Unified Lua and modules reply parsing and added RESP3 support to RM_Call (#9202)

## Current state
1. Lua has its own parser that handles parsing `reds.call` replies and translates them
  to Lua objects that can be used by the user Lua code. The parser partially handles
  resp3 (missing big number, verbatim, attribute, ...)
2. Modules have their own parser that handles parsing `RM_Call` replies and translates
  them to RedisModuleCallReply objects. The parser does not support resp3.

In addition, in the future, we want to add Redis Function (#8693) that will probably
support more languages. At some point maintaining so many parsers will stop
scaling (bug fixes and protocol changes will need to be applied on all of them).
We will probably end up with different parsers that support different parts of the
resp protocol (like we already have today with Lua and modules)

## PR Changes
This PR attempt to unified the reply parsing of Lua and modules (and in the future
Redis Function) by introducing a new parser unit (`resp_parser.c`). The new parser
handles parsing the reply and calls different callbacks to allow the users (another
unit that uses the parser, i.e, Lua, modules, or Redis Function) to analyze the reply.

### Lua API Additions
The code that handles reply parsing on `scripting.c` was removed. Instead, it uses
the resp_parser to parse and create a Lua object out of the reply. As mentioned
above the Lua parser did not handle parsing big numbers, verbatim, and attribute.
The new parser can handle those and so Lua also gets it for free.
Those are translated to Lua objects in the following way:
1. Big Number - Lua table `{'big_number':'<str representation for big number>'}`
2. Verbatim - Lua table `{'verbatim_string':{'format':'<verbatim format>', 'string':'<verbatim string value>'}}`
3. Attribute - currently ignored and not expose to the Lua parser, another issue will be open to decide how to expose it.

Tests were added to check resp3 reply parsing on Lua

### Modules API Additions
The reply parsing code on `module.c` was also removed and the new resp_parser is used instead.
In addition, the RedisModuleCallReply was also extracted to a separate unit located on `call_reply.c`
(in the future, this unit will also be used by Redis Function). A nice side effect of unified parsing is
that modules now also support resp3. Resp3 can be enabled by giving `3` as a parameter to the
fmt argument of `RM_Call`. It is also possible to give `0`, which will indicate an auto mode. i.e, Redis
will automatically chose the reply protocol base on the current client set on the RedisModuleCtx
(this mode will mostly be used when the module want to pass the reply to the client as is).
In addition, the following RedisModuleAPI were added to allow analyzing resp3 replies:

* New RedisModuleCallReply types:
   * `REDISMODULE_REPLY_MAP`
   * `REDISMODULE_REPLY_SET`
   * `REDISMODULE_REPLY_BOOL`
   * `REDISMODULE_REPLY_DOUBLE`
   * `REDISMODULE_REPLY_BIG_NUMBER`
   * `REDISMODULE_REPLY_VERBATIM_STRING`
   * `REDISMODULE_REPLY_ATTRIBUTE`

* New RedisModuleAPI:
   * `RedisModule_CallReplyDouble` - getting double value from resp3 double reply
   * `RedisModule_CallReplyBool` - getting boolean value from resp3 boolean reply
   * `RedisModule_CallReplyBigNumber` - getting big number value from resp3 big number reply
   * `RedisModule_CallReplyVerbatim` - getting format and value from resp3 verbatim reply
   * `RedisModule_CallReplySetElement` - getting element from resp3 set reply
   * `RedisModule_CallReplyMapElement` - getting key and value from resp3 map reply
   * `RedisModule_CallReplyAttribute` - getting a reply attribute
   * `RedisModule_CallReplyAttributeElement` - getting key and value from resp3 attribute reply
   
* New context flags:
   * `REDISMODULE_CTX_FLAGS_RESP3` - indicate that the client is using resp3

Tests were added to check the new RedisModuleAPI

### Modules API Changes
* RM_ReplyWithCallReply might return REDISMODULE_ERR if the given CallReply is in resp3
  but the client expects resp2. This is not a breaking change because in order to get a resp3
  CallReply one needs to specifically specify `3` as a parameter to the fmt argument of
  `RM_Call` (as mentioned above).

Tests were added to check this change

### More small Additions
* Added `debug set-disable-deny-scripts` that allows to turn on and off the commands no-script
flag protection. This is used by the Lua resp3 tests so it will be possible to run `debug protocol`
and check the resp3 parsing code.

Co-authored-by: Oran Agra <oran@redislabs.com>
Co-authored-by: Yossi Gottlieb <yossigo@gmail.com>

10 files changed, 1329 insertions, 290 deletions

diff --git a/src/Makefile b/src/Makefile
index 7ab9b6528..428e0bd34 100644
--- a/src/Makefile
+++ b/src/Makefile
@@ -282,7 +282,7 @@ endif
 
 REDIS_SERVER_NAME=redis-server$(PROG_SUFFIX)
 REDIS_SENTINEL_NAME=redis-sentinel$(PROG_SUFFIX)
-REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o memtest.o crcspeed.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o lolwut.o lolwut5.o lolwut6.o acl.o tracking.o connection.o tls.o sha256.o timeout.o setcpuaffinity.o monotonic.o mt19937-64.o
+REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o memtest.o crcspeed.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o lolwut.o lolwut5.o lolwut6.o acl.o tracking.o connection.o tls.o sha256.o timeout.o setcpuaffinity.o monotonic.o mt19937-64.o resp_parser.o call_reply.o
 REDIS_CLI_NAME=redis-cli$(PROG_SUFFIX)
 REDIS_CLI_OBJ=anet.o adlist.o dict.o redis-cli.o zmalloc.o release.o ae.o redisassert.o crcspeed.o crc64.o siphash.o crc16.o monotonic.o cli_common.o mt19937-64.o
 REDIS_BENCHMARK_NAME=redis-benchmark$(PROG_SUFFIX)
diff --git a/src/call_reply.c b/src/call_reply.c
new file mode 100644
index 000000000..5e7802463
--- /dev/null
+++ b/src/call_reply.c
@@ -0,0 +1,516 @@
+/*
+ * Copyright (c) 2009-2021, Redis Labs Ltd.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *   * Redistributions of source code must retain the above copyright notice,
+ *     this list of conditions and the following disclaimer.
+ *   * Redistributions in binary form must reproduce the above copyright
+ *     notice, this list of conditions and the following disclaimer in the
+ *     documentation and/or other materials provided with the distribution.
+ *   * Neither the name of Redis nor the names of its contributors may be used
+ *     to endorse or promote products derived from this software without
+ *     specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#include "server.h"
+#include "call_reply.h"
+
+#define REPLY_FLAG_ROOT (1<<0)
+#define REPLY_FLAG_PARSED (1<<1)
+#define REPLY_FLAG_RESP3 (1<<2)
+
+/* --------------------------------------------------------
+ * An opaque struct used to parse a RESP protocol reply and
+ * represent it. Used when parsing replies such as in RM_Call
+ * or Lua scripts.
+ * -------------------------------------------------------- */
+struct CallReply {
+    void *private_data;
+    sds original_proto; /* Available only for root reply. */
+    const char *proto;
+    size_t proto_len;
+    int type;       /* REPLY_... */
+    int flags;      /* REPLY_FLAG... */
+    size_t len;     /* Length of a string, or the number elements in an array. */
+    union {
+        const char *str; /* String pointer for string and error replies. This
+                          * does not need to be freed, always points inside
+                          * a reply->proto buffer of the reply object or, in
+                          * case of array elements, of parent reply objects. */
+        struct {
+            const char *str;
+            const char *format;
+        } verbatim_str;  /* Reply value for verbatim string */
+        long long ll;    /* Reply value for integer reply. */
+        double d;        /* Reply value for double reply. */
+        struct CallReply *array; /* Array of sub-reply elements. used for set, array, map, and attribute */
+    } val;
+
+    struct CallReply *attribute; /* attribute reply, NULL if not exists */
+};
+
+static void callReplySetSharedData(CallReply *rep, int type, const char *proto, size_t proto_len, int extra_flags) {
+    rep->type = type;
+    rep->proto = proto;
+    rep->proto_len = proto_len;
+    rep->flags |= extra_flags;
+}
+
+static void callReplyNull(void *ctx, const char *proto, size_t proto_len) {
+    CallReply *rep = ctx;
+    callReplySetSharedData(rep, REDISMODULE_REPLY_NULL, proto, proto_len, REPLY_FLAG_RESP3);
+}
+
+static void callReplyNullBulkString(void *ctx, const char *proto, size_t proto_len) {
+    CallReply *rep = ctx;
+    callReplySetSharedData(rep, REDISMODULE_REPLY_NULL, proto, proto_len, 0);
+}
+
+static void callReplyNullArray(void *ctx, const char *proto, size_t proto_len) {
+    CallReply *rep = ctx;
+    callReplySetSharedData(rep, REDISMODULE_REPLY_NULL, proto, proto_len, 0);
+}
+
+static void callReplyBulkString(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len) {
+    CallReply *rep = ctx;
+    callReplySetSharedData(rep, REDISMODULE_REPLY_STRING, proto, proto_len, 0);
+    rep->len = len;
+    rep->val.str = str;
+}
+
+static void callReplyError(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len) {
+    CallReply *rep = ctx;
+    callReplySetSharedData(rep, REDISMODULE_REPLY_ERROR, proto, proto_len, 0);
+    rep->len = len;
+    rep->val.str = str;
+}
+
+static void callReplySimpleStr(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len) {
+    CallReply *rep = ctx;
+    callReplySetSharedData(rep, REDISMODULE_REPLY_STRING, proto, proto_len, 0);
+    rep->len = len;
+    rep->val.str = str;
+}
+
+static void callReplyLong(void *ctx, long long val, const char *proto, size_t proto_len) {
+    CallReply *rep = ctx;
+    callReplySetSharedData(rep, REDISMODULE_REPLY_INTEGER, proto, proto_len, 0);
+    rep->val.ll = val;
+}
+
+static void callReplyDouble(void *ctx, double val, const char *proto, size_t proto_len) {
+    CallReply *rep = ctx;
+    callReplySetSharedData(rep, REDISMODULE_REPLY_DOUBLE, proto, proto_len, REPLY_FLAG_RESP3);
+    rep->val.d = val;
+}
+
+static void callReplyVerbatimString(void *ctx, const char *format, const char *str, size_t len, const char *proto, size_t proto_len) {
+    CallReply *rep = ctx;
+    callReplySetSharedData(rep, REDISMODULE_REPLY_VERBATIM_STRING, proto, proto_len, REPLY_FLAG_RESP3);
+    rep->len = len;
+    rep->val.verbatim_str.str = str;
+    rep->val.verbatim_str.format = format;
+}
+
+static void callReplyBigNumber(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len) {
+    CallReply *rep = ctx;
+    callReplySetSharedData(rep, REDISMODULE_REPLY_BIG_NUMBER, proto, proto_len, REPLY_FLAG_RESP3);
+    rep->len = len;
+    rep->val.str = str;
+}
+
+static void callReplyBool(void *ctx, int val, const char *proto, size_t proto_len) {
+    CallReply *rep = ctx;
+    callReplySetSharedData(rep, REDISMODULE_REPLY_BOOL, proto, proto_len, REPLY_FLAG_RESP3);
+    rep->val.ll = val;
+}
+
+static void callReplyParseCollection(ReplyParser *parser, CallReply *rep, size_t len, const char *proto, size_t elements_per_entry) {
+    rep->len = len;
+    rep->val.array = zcalloc(elements_per_entry * len * sizeof(CallReply));
+    for (size_t i = 0; i < len * elements_per_entry; i += elements_per_entry) {
+        for (size_t j = 0 ; j < elements_per_entry ; ++j) {
+            parseReply(parser, rep->val.array + i + j);
+            rep->val.array[i + j].flags |= REPLY_FLAG_PARSED;
+            rep->val.array[i + j].private_data = rep->private_data;
+            if (rep->val.array[i + j].flags & REPLY_FLAG_RESP3) {
+                /* If one of the sub-replies is RESP3, then the current reply is also RESP3. */
+                rep->flags |= REPLY_FLAG_RESP3;
+            }
+        }
+    }
+    rep->proto = proto;
+    rep->proto_len = parser->curr_location - proto;
+}
+
+static void callReplyAttribute(ReplyParser *parser, void *ctx, size_t len, const char *proto) {
+    CallReply *rep = ctx;
+    rep->attribute = zcalloc(sizeof(CallReply));
+
+    /* Continue parsing the attribute reply */
+    rep->attribute->len = len;
+    rep->attribute->type = REDISMODULE_REPLY_ATTRIBUTE;
+    callReplyParseCollection(parser, rep->attribute, len, proto, 2);
+    rep->attribute->flags |= REPLY_FLAG_PARSED | REPLY_FLAG_RESP3;
+    rep->attribute->private_data = rep->private_data;
+
+    /* Continue parsing the reply */
+    parseReply(parser, rep);
+
+    /* In this case we need to fix the proto address and len, it should start from the attribute */
+    rep->proto = proto;
+    rep->proto_len = parser->curr_location - proto;
+    rep->flags |= REPLY_FLAG_RESP3;
+}
+
+static void callReplyArray(ReplyParser *parser, void *ctx, size_t len, const char *proto) {
+    CallReply *rep = ctx;
+    rep->type = REDISMODULE_REPLY_ARRAY;
+    callReplyParseCollection(parser, rep, len, proto, 1);
+}
+
+static void callReplySet(ReplyParser *parser, void *ctx, size_t len, const char *proto) {
+    CallReply *rep = ctx;
+    rep->type = REDISMODULE_REPLY_SET;
+    callReplyParseCollection(parser, rep, len, proto, 1);
+    rep->flags |= REPLY_FLAG_RESP3;
+}
+
+static void callReplyMap(ReplyParser *parser, void *ctx, size_t len, const char *proto) {
+    CallReply *rep = ctx;
+    rep->type = REDISMODULE_REPLY_MAP;
+    callReplyParseCollection(parser, rep, len, proto, 2);
+    rep->flags |= REPLY_FLAG_RESP3;
+}
+
+static void callReplyParseError(void *ctx) {
+    CallReply *rep = ctx;
+    rep->type = REDISMODULE_REPLY_UNKNOWN;
+}
+
+/* Recursively free the current call reply and its sub-replies. */
+static void freeCallReplyInternal(CallReply *rep) {
+    if (rep->type == REDISMODULE_REPLY_ARRAY || rep->type == REDISMODULE_REPLY_SET) {
+        for (size_t i = 0 ; i < rep->len ; ++i) {
+            freeCallReplyInternal(rep->val.array + i);
+        }
+        zfree(rep->val.array);
+    }
+
+    if (rep->type == REDISMODULE_REPLY_MAP || rep->type == REDISMODULE_REPLY_ATTRIBUTE) {
+        for (size_t i = 0 ; i < rep->len ; ++i) {
+            freeCallReplyInternal(rep->val.array + i * 2);
+            freeCallReplyInternal(rep->val.array + i * 2 + 1);
+        }
+        zfree(rep->val.array);
+    }
+
+    if (rep->attribute) {
+        freeCallReplyInternal(rep->attribute);
+        zfree(rep->attribute);
+    }
+}
+
+/* Free the given call reply and its children (in case of nested reply) recursively.
+ * If private data was set when the CallReply was created it will not be freed, as it's
+ * the caller's responsibility to free it before calling freeCallReply(). */
+void freeCallReply(CallReply *rep) {
+    if (!(rep->flags & REPLY_FLAG_ROOT)) {
+        return;
+    }
+    if (rep->flags & REPLY_FLAG_PARSED) {
+        freeCallReplyInternal(rep);
+    }
+    sdsfree(rep->original_proto);
+    zfree(rep);
+}
+
+static const ReplyParserCallbacks DefaultParserCallbacks = {
+    .null_callback = callReplyNull,
+    .bulk_string_callback = callReplyBulkString,
+    .null_bulk_string_callback = callReplyNullBulkString,
+    .null_array_callback = callReplyNullArray,
+    .error_callback = callReplyError,
+    .simple_str_callback = callReplySimpleStr,
+    .long_callback = callReplyLong,
+    .array_callback = callReplyArray,
+    .set_callback = callReplySet,
+    .map_callback = callReplyMap,
+    .double_callback = callReplyDouble,
+    .bool_callback = callReplyBool,
+    .big_number_callback = callReplyBigNumber,
+    .verbatim_string_callback = callReplyVerbatimString,
+    .attribute_callback = callReplyAttribute,
+    .error = callReplyParseError,
+};
+
+/* Parse the buffer located in rep->original_proto and update the CallReply
+ * structure to represent its contents. */
+static void callReplyParse(CallReply *rep) {
+    if (rep->flags & REPLY_FLAG_PARSED) {
+        return;
+    }
+
+    ReplyParser parser = {.curr_location = rep->proto, .callbacks = DefaultParserCallbacks};
+
+    parseReply(&parser, rep);
+    rep->flags |= REPLY_FLAG_PARSED;
+}
+
+/* Return the call reply type (REDISMODULE_REPLY_...). */
+int callReplyType(CallReply *rep) {
+    if (!rep) return REDISMODULE_REPLY_UNKNOWN;
+    callReplyParse(rep);
+    return rep->type;
+}
+
+/* Return reply string as buffer and len. Applicable to:
+ * - REDISMODULE_REPLY_STRING
+ * - REDISMODULE_REPLY_ERROR
+ *
+ * The return value is borrowed from CallReply, so it must not be freed
+ * explicitly or used after CallReply itself is freed.
+ *
+ * The returned value is not NULL terminated and its length is returned by
+ * reference through len, which must not be NULL.
+ */
+const char *callReplyGetString(CallReply *rep, size_t *len) {
+    callReplyParse(rep);
+    if (rep->type != REDISMODULE_REPLY_STRING &&
+        rep->type != REDISMODULE_REPLY_ERROR) return NULL;
+    if (len) *len = rep->len;
+    return rep->val.str;
+}
+
+/* Return a long long reply value. Applicable to:
+ * - REDISMODULE_REPLY_INTEGER
+ */
+long long callReplyGetLongLong(CallReply *rep) {
+    callReplyParse(rep);
+    if (rep->type != REDISMODULE_REPLY_INTEGER) return LLONG_MIN;
+    return rep->val.ll;
+}
+
+/* Return a double reply value. Applicable to:
+ * - REDISMODULE_REPLY_DOUBLE
+ */
+double callReplyGetDouble(CallReply *rep) {
+    callReplyParse(rep);
+    if (rep->type != REDISMODULE_REPLY_DOUBLE) return LLONG_MIN;
+    return rep->val.d;
+}
+
+/* Return a reply Boolean value. Applicable to:
+ * - REDISMODULE_REPLY_BOOL
+ */
+int callReplyGetBool(CallReply *rep) {
+    callReplyParse(rep);
+    if (rep->type != REDISMODULE_REPLY_BOOL) return INT_MIN;
+    return rep->val.ll;
+}
+
+/* Return reply length. Applicable to:
+ * - REDISMODULE_REPLY_STRING
+ * - REDISMODULE_REPLY_ERROR
+ * - REDISMODULE_REPLY_ARRAY
+ * - REDISMODULE_REPLY_SET
+ * - REDISMODULE_REPLY_MAP
+ * - REDISMODULE_REPLY_ATTRIBUTE
+ */
+size_t callReplyGetLen(CallReply *rep) {
+    callReplyParse(rep);
+    switch(rep->type) {
+        case REDISMODULE_REPLY_STRING:
+        case REDISMODULE_REPLY_ERROR:
+        case REDISMODULE_REPLY_ARRAY:
+        case REDISMODULE_REPLY_SET:
+        case REDISMODULE_REPLY_MAP:
+        case REDISMODULE_REPLY_ATTRIBUTE:
+            return rep->len;
+        default:
+            return 0;
+    }
+}
+
+static CallReply *callReplyGetCollectionElement(CallReply *rep, size_t idx, int elements_per_entry) {
+    if (idx >= rep->len * elements_per_entry) return NULL; // real len is rep->len * elements_per_entry
+    return rep->val.array+idx;
+}
+
+/* Return a reply array element at a given index. Applicable to:
+ * - REDISMODULE_REPLY_ARRAY
+ *
+ * The return value is borrowed from CallReply, so it must not be freed
+ * explicitly or used after CallReply itself is freed.
+ */
+CallReply *callReplyGetArrayElement(CallReply *rep, size_t idx) {
+    callReplyParse(rep);
+    if (rep->type != REDISMODULE_REPLY_ARRAY) return NULL;
+    return callReplyGetCollectionElement(rep, idx, 1);
+}
+
+/* Return a reply set element at a given index. Applicable to:
+ * - REDISMODULE_REPLY_SET
+ *
+ * The return value is borrowed from CallReply, so it must not be freed
+ * explicitly or used after CallReply itself is freed.
+ */
+CallReply *callReplyGetSetElement(CallReply *rep, size_t idx) {
+    callReplyParse(rep);
+    if (rep->type != REDISMODULE_REPLY_SET) return NULL;
+    return callReplyGetCollectionElement(rep, idx, 1);
+}
+
+static int callReplyGetMapElementInternal(CallReply *rep, size_t idx, CallReply **key, CallReply **val, int type) {
+    callReplyParse(rep);
+    if (rep->type != type) return C_ERR;
+    if (idx >= rep->len) return C_ERR;
+    if (key) *key = callReplyGetCollectionElement(rep, idx * 2, 2);
+    if (val) *val = callReplyGetCollectionElement(rep, idx * 2 + 1, 2);
+    return C_OK;
+}
+
+/* Retrieve a map reply key and value at a given index. Applicable to:
+ * - REDISMODULE_REPLY_MAP
+ *
+ * The key and value are returned by reference through key and val,
+ * which may also be NULL if not needed.
+ *
+ * Returns C_OK on success or C_ERR if reply type mismatches, or if idx is out
+ * of range.
+ *
+ * The returned values are borrowed from CallReply, so they must not be freed
+ * explicitly or used after CallReply itself is freed.
+ */
+int callReplyGetMapElement(CallReply *rep, size_t idx, CallReply **key, CallReply **val) {
+    return callReplyGetMapElementInternal(rep, idx, key, val, REDISMODULE_REPLY_MAP);
+}
+
+/* Return reply attribute, or NULL if it does not exist. Applicable to all replies.
+ *
+ * The returned values are borrowed from CallReply, so they must not be freed
+ * explicitly or used after CallReply itself is freed.
+ */
+CallReply *callReplyGetAttribute(CallReply *rep) {
+    return rep->attribute;
+}
+
+/* Retrieve attribute reply key and value at a given index. Applicable to:
+ * - REDISMODULE_REPLY_ATTRIBUTE
+ *
+ * The key and value are returned by reference through key and val,
+ * which may also be NULL if not needed.
+ *
+ * Returns C_OK on success or C_ERR if reply type mismatches, or if idx is out
+ * of range.
+ *
+ * The returned values are borrowed from CallReply, so they must not be freed
+ * explicitly or used after CallReply itself is freed.
+ */
+int callReplyGetAttributeElement(CallReply *rep, size_t idx, CallReply **key, CallReply **val) {
+    return callReplyGetMapElementInternal(rep, idx, key, val, REDISMODULE_REPLY_MAP);
+}
+
+/* Return a big number reply value. Applicable to:
+ * - REDISMODULE_REPLY_BIG_NUMBER
+ *
+ * The returned values are borrowed from CallReply, so they must not be freed
+ * explicitly or used after CallReply itself is freed.
+ *
+ * The return value is guaranteed to be a big number, as described in the RESP3
+ * protocol specifications.
+ *
+ * The returned value is not NULL terminated and its length is returned by
+ * reference through len, which must not be NULL.
+ */
+const char *callReplyGetBigNumber(CallReply *rep, size_t *len) {
+    callReplyParse(rep);
+    if (rep->type != REDISMODULE_REPLY_BIG_NUMBER) return NULL;
+    *len = rep->len;
+    return rep->val.str;
+}
+
+/* Return a verbatim string reply value. Applicable to:
+ * - REDISMODULE_REPLY_VERBATIM_STRING
+ *
+ * If format is non-NULL, the verbatim reply format is also returned by value.
+ *
+ * The optional output argument can be given to get a verbatim reply
+ * format, or can be set NULL if not needed.
+ *
+ * The return value is borrowed from CallReply, so it must not be freed
+ * explicitly or used after CallReply itself is freed.
+ *
+ * The returned value is not NULL terminated and its length is returned by
+ * reference through len, which must not be NULL.
+ */
+const char *callReplyGetVerbatim(CallReply *rep, size_t *len, const char **format){
+    callReplyParse(rep);
+    if (rep->type != REDISMODULE_REPLY_VERBATIM_STRING) return NULL;
+    *len = rep->len;
+    if (format) *format = rep->val.verbatim_str.format;
+    return rep->val.verbatim_str.str;
+}
+
+/* Return the current reply blob.
+ *
+ * The return value is borrowed from CallReply, so it must not be freed
+ * explicitly or used after CallReply itself is freed.
+ */
+const char *callReplyGetProto(CallReply *rep, size_t *proto_len) {
+    *proto_len = rep->proto_len;
+    return rep->proto;
+}
+
+/* Return CallReply private data, as set by the caller on callReplyCreate().
+ */
+void *callReplyGetPrivateData(CallReply *rep) {
+    return rep->private_data;
+}
+
+/* Return true if the reply or one of it sub-replies is RESP3 formatted. */
+int callReplyIsResp3(CallReply *rep) {
+    return rep->flags & REPLY_FLAG_RESP3;
+}
+
+/* Create a new CallReply struct from the reply blob.
+ *
+ * The function will own the reply blob, so it must not be used or freed by
+ * the caller after passing it to this function.
+ *
+ * The reply blob will be freed when the returned CallReply struct is later
+ * freed using freeCallReply().
+ *
+ * The private_data is optional and can later be accessed using
+ * callReplyGetPrivateData().
+ *
+ * NOTE: The parser used for parsing the reply and producing CallReply is
+ * designed to handle valid replies created by Redis itself. IT IS NOT
+ * DESIGNED TO HANDLE USER INPUT and using it to parse invalid replies is
+ * unsafe.
+ */
+CallReply *callReplyCreate(sds reply, void *private_data) {
+    CallReply *res = zmalloc(sizeof(*res));
+    res->flags = REPLY_FLAG_ROOT;
+    res->original_proto = reply;
+    res->proto = reply;
+    res->proto_len = sdslen(reply);
+    res->private_data = private_data;
+    res->attribute = NULL;
+    return res;
+}
diff --git a/src/call_reply.h b/src/call_reply.h
new file mode 100644
index 000000000..5b07dc437
--- /dev/null
+++ b/src/call_reply.h
@@ -0,0 +1,56 @@
+/*
+ * Copyright (c) 2009-2021, Redis Labs Ltd.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *   * Redistributions of source code must retain the above copyright notice,
+ *     this list of conditions and the following disclaimer.
+ *   * Redistributions in binary form must reproduce the above copyright
+ *     notice, this list of conditions and the following disclaimer in the
+ *     documentation and/or other materials provided with the distribution.
+ *   * Neither the name of Redis nor the names of its contributors may be used
+ *     to endorse or promote products derived from this software without
+ *     specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#ifndef SRC_CALL_REPLY_H_
+#define SRC_CALL_REPLY_H_
+
+#include "resp_parser.h"
+
+typedef struct CallReply CallReply;
+
+CallReply *callReplyCreate(sds reply, void *private_data);
+int callReplyType(CallReply *rep);
+const char *callReplyGetString(CallReply *rep, size_t *len);
+long long callReplyGetLongLong(CallReply *rep);
+double callReplyGetDouble(CallReply *rep);
+int callReplyGetBool(CallReply *rep);
+size_t callReplyGetLen(CallReply *rep);
+CallReply *callReplyGetArrayElement(CallReply *rep, size_t idx);
+CallReply *callReplyGetSetElement(CallReply *rep, size_t idx);
+int callReplyGetMapElement(CallReply *rep, size_t idx, CallReply **key, CallReply **val);
+CallReply *callReplyGetAttribute(CallReply *rep);
+int callReplyGetAttributeElement(CallReply *rep, size_t idx, CallReply **key, CallReply **val);
+const char *callReplyGetBigNumber(CallReply *rep, size_t *len);
+const char *callReplyGetVerbatim(CallReply *rep, size_t *len, const char **format);
+const char *callReplyGetProto(CallReply *rep, size_t *len);
+void *callReplyGetPrivateData(CallReply *rep);
+int callReplyIsResp3(CallReply *rep);
+void freeCallReply(CallReply *rep);
+
+#endif /* SRC_CALL_REPLY_H_ */
diff --git a/src/debug.c b/src/debug.c
index d3af4db66..9598e6c6c 100644
--- a/src/debug.c
+++ b/src/debug.c
@@ -873,6 +873,10 @@ NULL
     {
         stringmatchlen_fuzz_test();
         addReplyStatus(c,"Apparently Redis did not crash: test passed");
+    } else if (!strcasecmp(c->argv[1]->ptr,"set-disable-deny-scripts") && c->argc == 3)
+    {
+        server.lua_disable_deny_script = atoi(c->argv[2]->ptr);;
+        addReply(c,shared.ok);
     } else if (!strcasecmp(c->argv[1]->ptr,"config-rewrite-force-all") && c->argc == 2)
     {
         if (rewriteConfig(server.configfile, 1) == -1)
diff --git a/src/module.c b/src/module.c
index 51b4d8dd6..f8d5e21d4 100644
--- a/src/module.c
+++ b/src/module.c
@@ -56,6 +56,7 @@
 #include "slowlog.h"
 #include "rdb.h"
 #include "monotonic.h"
+#include "call_reply.h"
 #include <dlfcn.h>
 #include <sys/stat.h>
 #include <sys/wait.h>
@@ -232,22 +233,7 @@ typedef struct RedisModuleCommandProxy RedisModuleCommandProxy;
 /* Reply of RM_Call() function. The function is filled in a lazy
  * way depending on the function called on the reply structure. By default
  * only the type, proto and protolen are filled. */
-typedef struct RedisModuleCallReply {
-    RedisModuleCtx *ctx;
-    int type;       /* REDISMODULE_REPLY_... */
-    int flags;      /* REDISMODULE_REPLYFLAG_...  */
-    size_t len;     /* Len of strings or num of elements of arrays. */
-    char *proto;    /* Raw reply protocol. An SDS string at top-level object. */
-    size_t protolen;/* Length of protocol. */
-    union {
-        const char *str; /* String pointer for string and error replies. This
-                            does not need to be freed, always points inside
-                            a reply->proto buffer of the reply object or, in
-                            case of array elements, of parent reply objects. */
-        long long ll;    /* Reply value for integer reply. */
-        struct RedisModuleCallReply *array; /* Array of sub-reply elements. */
-    } val;
-} RedisModuleCallReply;
+typedef struct CallReply RedisModuleCallReply;
 
 /* Structure representing a blocked client. We get a pointer to such
  * an object when blocking from modules. */
@@ -350,6 +336,8 @@ typedef struct RedisModuleServerInfoData {
 #define REDISMODULE_ARGV_REPLICATE (1<<0)
 #define REDISMODULE_ARGV_NO_AOF (1<<1)
 #define REDISMODULE_ARGV_NO_REPLICAS (1<<2)
+#define REDISMODULE_ARGV_RESP_3 (1<<3)
+#define REDISMODULE_ARGV_RESP_AUTO (1<<4)
 
 /* Determine whether Redis should signalModifiedKey implicitly.
  * In case 'ctx' has no 'module' member (and therefore no module->options),
@@ -1851,12 +1839,25 @@ int RM_ReplyWithBool(RedisModuleCtx *ctx, int b) {
  * execute some command, as we want to reply to the client exactly the
  * same reply we obtained by the command.
  *
- * The function always returns REDISMODULE_OK. */
+ * Return:
+ * - REDISMODULE_OK on success.
+ * - REDISMODULE_ERR if the given reply is in RESP3 format but the client expects RESP2.
+ *   In case of an error, it's the module writer responsibility to translate the reply
+ *   to RESP2 (or handle it differently by returning an error). Notice that for
+ *   module writer convenience, it is possible to pass `0` as a parameter to the fmt
+ *   argument of `RM_Call` so that the RedisModuleCallReply will return in the same
+ *   protocol (RESP2 or RESP3) as set in the current client's context. */
 int RM_ReplyWithCallReply(RedisModuleCtx *ctx, RedisModuleCallReply *reply) {
     client *c = moduleGetReplyClient(ctx);
     if (c == NULL) return REDISMODULE_OK;
-    sds proto = sdsnewlen(reply->proto, reply->protolen);
-    addReplySds(c,proto);
+    if (c->resp == 2 && callReplyIsResp3(reply)) {
+        /* The reply is in RESP3 format and the client is RESP2,
+         * so it isn't possible to send this reply to the client. */
+        return REDISMODULE_ERR;
+    }
+    size_t proto_len;
+    const char *proto = callReplyGetProto(reply, &proto_len);
+    addReplyProto(c, proto, proto_len);
     return REDISMODULE_OK;
 }
 
@@ -2247,6 +2248,9 @@ int RM_GetSelectedDb(RedisModuleCtx *ctx) {
  *
  *  * REDISMODULE_CTX_FLAGS_IS_CHILD: Redis is currently running inside
  *                                    background child process.
+ *
+ *  * REDISMODULE_CTX_FLAGS_RESP3: Indicate the that client attached to this
+ *                                 context is using RESP3.
  */
 int RM_GetContextFlags(RedisModuleCtx *ctx) {
     int flags = 0;
@@ -2259,6 +2263,9 @@ int RM_GetContextFlags(RedisModuleCtx *ctx) {
             /* Module command received from MASTER, is replicated. */
             if (ctx->client->flags & CLIENT_MASTER)
                 flags |= REDISMODULE_CTX_FLAGS_REPLICATED;
+            if (ctx->client->resp == 3) {
+                flags |= REDISMODULE_CTX_FLAGS_RESP3;
+            }
         }
 
         /* For DIRTY flags, we need the blocked client if used */
@@ -3922,141 +3929,14 @@ long long RM_StreamTrimByID(RedisModuleKey *key, int flags, RedisModuleStreamID
  * RM_Call() sends a command to Redis. The remaining functions handle the reply.
  * -------------------------------------------------------------------------- */
 
-/* Create a new RedisModuleCallReply object. The processing of the reply
- * is lazy, the object is just populated with the raw protocol and later
- * is processed as needed. Initially we just make sure to set the right
- * reply type, which is extremely cheap to do. */
-RedisModuleCallReply *moduleCreateCallReplyFromProto(RedisModuleCtx *ctx, sds proto) {
-    RedisModuleCallReply *reply = zmalloc(sizeof(*reply));
-    reply->ctx = ctx;
-    reply->proto = proto;
-    reply->protolen = sdslen(proto);
-    reply->flags = REDISMODULE_REPLYFLAG_TOPARSE; /* Lazy parsing. */
-    switch(proto[0]) {
-    case '$':
-    case '+': reply->type = REDISMODULE_REPLY_STRING; break;
-    case '-': reply->type = REDISMODULE_REPLY_ERROR; break;
-    case ':': reply->type = REDISMODULE_REPLY_INTEGER; break;
-    case '*': reply->type = REDISMODULE_REPLY_ARRAY; break;
-    default: reply->type = REDISMODULE_REPLY_UNKNOWN; break;
-    }
-    if ((proto[0] == '*' || proto[0] == '$') && proto[1] == '-')
-        reply->type = REDISMODULE_REPLY_NULL;
-    return reply;
-}
 
 void moduleParseCallReply_Int(RedisModuleCallReply *reply);
 void moduleParseCallReply_BulkString(RedisModuleCallReply *reply);
 void moduleParseCallReply_SimpleString(RedisModuleCallReply *reply);
 void moduleParseCallReply_Array(RedisModuleCallReply *reply);
 
-/* Do nothing if REDISMODULE_REPLYFLAG_TOPARSE is false, otherwise
- * use the protocol of the reply in reply->proto in order to fill the
- * reply with parsed data according to the reply type. */
-void moduleParseCallReply(RedisModuleCallReply *reply) {
-    if (!(reply->flags & REDISMODULE_REPLYFLAG_TOPARSE)) return;
-    reply->flags &= ~REDISMODULE_REPLYFLAG_TOPARSE;
-
-    switch(reply->proto[0]) {
-    case ':': moduleParseCallReply_Int(reply); break;
-    case '$': moduleParseCallReply_BulkString(reply); break;
-    case '-': /* handled by next item. */
-    case '+': moduleParseCallReply_SimpleString(reply); break;
-    case '*': moduleParseCallReply_Array(reply); break;
-    }
-}
-
-void moduleParseCallReply_Int(RedisModuleCallReply *reply) {
-    char *proto = reply->proto;
-    char *p = strchr(proto+1,'\r');
-
-    string2ll(proto+1,p-proto-1,&reply->val.ll);
-    reply->protolen = p-proto+2;
-    reply->type = REDISMODULE_REPLY_INTEGER;
-}
-
-void moduleParseCallReply_BulkString(RedisModuleCallReply *reply) {
-    char *proto = reply->proto;
-    char *p = strchr(proto+1,'\r');
-    long long bulklen;
-
-    string2ll(proto+1,p-proto-1,&bulklen);
-    if (bulklen == -1) {
-        reply->protolen = p-proto+2;
-        reply->type = REDISMODULE_REPLY_NULL;
-    } else {
-        reply->val.str = p+2;
-        reply->len = bulklen;
-        reply->protolen = p-proto+2+bulklen+2;
-        reply->type = REDISMODULE_REPLY_STRING;
-    }
-}
 
-void moduleParseCallReply_SimpleString(RedisModuleCallReply *reply) {
-    char *proto = reply->proto;
-    char *p = strchr(proto+1,'\r');
-
-    reply->val.str = proto+1;
-    reply->len = p-proto-1;
-    reply->protolen = p-proto+2;
-    reply->type = proto[0] == '+' ? REDISMODULE_REPLY_STRING :
-                                    REDISMODULE_REPLY_ERROR;
-}
 
-void moduleParseCallReply_Array(RedisModuleCallReply *reply) {
-    char *proto = reply->proto;
-    char *p = strchr(proto+1,'\r');
-    long long arraylen, j;
-
-    string2ll(proto+1,p-proto-1,&arraylen);
-    p += 2;
-
-    if (arraylen == -1) {
-        reply->protolen = p-proto;
-        reply->type = REDISMODULE_REPLY_NULL;
-        return;
-    }
-
-    reply->val.array = zmalloc(sizeof(RedisModuleCallReply)*arraylen);
-    reply->len = arraylen;
-    for (j = 0; j < arraylen; j++) {
-        RedisModuleCallReply *ele = reply->val.array+j;
-        ele->flags = REDISMODULE_REPLYFLAG_NESTED |
-                     REDISMODULE_REPLYFLAG_TOPARSE;
-        ele->proto = p;
-        ele->ctx = reply->ctx;
-        moduleParseCallReply(ele);
-        p += ele->protolen;
-    }
-    reply->protolen = p-proto;
-    reply->type = REDISMODULE_REPLY_ARRAY;
-}
-
-/* Recursive free reply function. */
-void moduleFreeCallReplyRec(RedisModuleCallReply *reply, int freenested){
-    /* Don't free nested replies by default: the user must always free the
-     * toplevel reply. However be gentle and don't crash if the module
-     * misuses the API. */
-    if (!freenested && reply->flags & REDISMODULE_REPLYFLAG_NESTED) return;
-
-    if (!(reply->flags & REDISMODULE_REPLYFLAG_TOPARSE)) {
-        if (reply->type == REDISMODULE_REPLY_ARRAY) {
-            size_t j;
-            for (j = 0; j < reply->len; j++)
-                moduleFreeCallReplyRec(reply->val.array+j,1);
-            zfree(reply->val.array);
-        }
-    }
-
-    /* For nested replies, we don't free reply->proto (which if not NULL
-     * references the parent reply->proto buffer), nor the structure
-     * itself which is allocated as an array of structures, and is freed
-     * when the array value is released. */
-    if (!(reply->flags & REDISMODULE_REPLYFLAG_NESTED)) {
-        if (reply->proto) sdsfree(reply->proto);
-        zfree(reply);
-    }
-}
 
 /* Free a Call reply and all the nested replies it contains if it's an
  * array. */
@@ -4064,69 +3944,133 @@ void RM_FreeCallReply(RedisModuleCallReply *reply) {
     /* This is a wrapper for the recursive free reply function. This is needed
      * in order to have the first level function to return on nested replies,
      * but only if called by the module API. */
-    RedisModuleCtx *ctx = reply->ctx;
-    moduleFreeCallReplyRec(reply,0);
+    RedisModuleCtx *ctx = callReplyGetPrivateData(reply);
+    freeCallReply(reply);
     autoMemoryFreed(ctx,REDISMODULE_AM_REPLY,reply);
 }
 
-/* Return the reply type. */
+/* Return the reply type as one of the following:
+ *
+ * - REDISMODULE_REPLY_UNKNOWN
+ * - REDISMODULE_REPLY_STRING
+ * - REDISMODULE_REPLY_ERROR
+ * - REDISMODULE_REPLY_INTEGER
+ * - REDISMODULE_REPLY_ARRAY
+ * - REDISMODULE_REPLY_NULL
+ * - REDISMODULE_REPLY_MAP
+ * - REDISMODULE_REPLY_SET
+ * - REDISMODULE_REPLY_BOOL
+ * - REDISMODULE_REPLY_DOUBLE
+ * - REDISMODULE_REPLY_BIG_NUMBER
+ * - REDISMODULE_REPLY_VERBATIM_STRING
+ * - REDISMODULE_REPLY_ATTRIBUTE */
 int RM_CallReplyType(RedisModuleCallReply *reply) {
-    if (!reply) return REDISMODULE_REPLY_UNKNOWN;
-    return reply->type;
+    return callReplyType(reply);
 }
 
 /* Return the reply type length, where applicable. */
 size_t RM_CallReplyLength(RedisModuleCallReply *reply) {
-    moduleParseCallReply(reply);
-    switch(reply->type) {
-    case REDISMODULE_REPLY_STRING:
-    case REDISMODULE_REPLY_ERROR:
-    case REDISMODULE_REPLY_ARRAY:
-        return reply->len;
-    default:
-        return 0;
-    }
+    return callReplyGetLen(reply);
 }
 
 /* Return the 'idx'-th nested call reply element of an array reply, or NULL
  * if the reply type is wrong or the index is out of range. */
 RedisModuleCallReply *RM_CallReplyArrayElement(RedisModuleCallReply *reply, size_t idx) {
-    moduleParseCallReply(reply);
-    if (reply->type != REDISMODULE_REPLY_ARRAY) return NULL;
-    if (idx >= reply->len) return NULL;
-    return reply->val.array+idx;
+    return callReplyGetArrayElement(reply, idx);
 }
 
 /* Return the long long of an integer reply. */
 long long RM_CallReplyInteger(RedisModuleCallReply *reply) {
-    moduleParseCallReply(reply);
-    if (reply->type != REDISMODULE_REPLY_INTEGER) return LLONG_MIN;
-    return reply->val.ll;
+    return callReplyGetLongLong(reply);
+}
+
+/* Return the double value of a double reply. */
+double RM_CallReplyDouble(RedisModuleCallReply *reply) {
+    return callReplyGetDouble(reply);
+}
+
+/* Return the big number value of a big number reply. */
+const char *RM_CallReplyBigNumber(RedisModuleCallReply *reply, size_t *len) {
+    return callReplyGetBigNumber(reply, len);
+}
+
+/* Return the value of an verbatim string reply,
+ * An optional output argument can be given to get verbatim reply format. */
+const char *RM_CallReplyVerbatim(RedisModuleCallReply *reply, size_t *len, const char **format) {
+    return callReplyGetVerbatim(reply, len, format);
+}
+
+/* Return the Boolean value of a Boolean reply. */
+int RM_CallReplyBool(RedisModuleCallReply *reply) {
+    return callReplyGetBool(reply);
+}
+
+/* Return the 'idx'-th nested call reply element of a set reply, or NULL
+ * if the reply type is wrong or the index is out of range. */
+RedisModuleCallReply *RM_CallReplySetElement(RedisModuleCallReply *reply, size_t idx) {
+    return callReplyGetSetElement(reply, idx);
+}
+
+/* Retrieve the 'idx'-th key and value of a map reply.
+ *
+ * Returns:
+ * - REDISMODULE_OK on success.
+ * - REDISMODULE_ERR if idx out of range or if the reply type is wrong.
+ *
+ * The `key` and `value` arguments are used to return by reference, and may be
+ * NULL if not required. */
+int RM_CallReplyMapElement(RedisModuleCallReply *reply, size_t idx, RedisModuleCallReply **key, RedisModuleCallReply **val) {
+    if (callReplyGetMapElement(reply, idx, key, val) == C_OK){
+        return REDISMODULE_OK;
+    }
+    return REDISMODULE_ERR;
+}
+
+/* Return the attribute of the given reply, or NULL if no attribute exists. */
+RedisModuleCallReply *RM_CallReplyAttribute(RedisModuleCallReply *reply) {
+    return callReplyGetAttribute(reply);
+}
+
+/* Retrieve the 'idx'-th key and value of a attribute reply.
+ *
+ * Returns:
+ * - REDISMODULE_OK on success.
+ * - REDISMODULE_ERR if idx out of range or if the reply type is wrong.
+ *
+ * The `key` and `value` arguments are used to return by reference, and may be
+ * NULL if not required. */
+int RM_CallReplyAttributeElement(RedisModuleCallReply *reply, size_t idx, RedisModuleCallReply **key, RedisModuleCallReply **val) {
+    if (callReplyGetAttributeElement(reply, idx, key, val) == C_OK){
+        return REDISMODULE_OK;
+    }
+    return REDISMODULE_ERR;
 }
 
 /* Return the pointer and length of a string or error reply. */
 const char *RM_CallReplyStringPtr(RedisModuleCallReply *reply, size_t *len) {
-    moduleParseCallReply(reply);
-    if (reply->type != REDISMODULE_REPLY_STRING &&
-        reply->type != REDISMODULE_REPLY_ERROR) return NULL;
-    if (len) *len = reply->len;
-    return reply->val.str;
+    size_t private_len;
+    if (!len) len = &private_len;
+    return callReplyGetString(reply, len);
 }
 
 /* Return a new string object from a call reply of type string, error or
  * integer. Otherwise (wrong reply type) return NULL. */
 RedisModuleString *RM_CreateStringFromCallReply(RedisModuleCallReply *reply) {
-    moduleParseCallReply(reply);
-    switch(reply->type) {
-    case REDISMODULE_REPLY_STRING:
-    case REDISMODULE_REPLY_ERROR:
-        return RM_CreateString(reply->ctx,reply->val.str,reply->len);
-    case REDISMODULE_REPLY_INTEGER: {
-        char buf[64];
-        int len = ll2string(buf,sizeof(buf),reply->val.ll);
-        return RM_CreateString(reply->ctx,buf,len);
-        }
-    default: return NULL;
+    RedisModuleCtx* ctx = callReplyGetPrivateData(reply);
+    size_t len;
+    const char *str;
+    switch(callReplyType(reply)) {
+        case REDISMODULE_REPLY_STRING:
+        case REDISMODULE_REPLY_ERROR:
+            str = callReplyGetString(reply, &len);
+            return RM_CreateString(ctx, str, len);
+        case REDISMODULE_REPLY_INTEGER: {
+            char buf[64];
+            int len = ll2string(buf,sizeof(buf),callReplyGetLongLong(reply));
+            return RM_CreateString(ctx ,buf,len);
+            }
+        default:
+            return NULL;
     }
 }
 
@@ -4140,6 +4084,8 @@ RedisModuleString *RM_CreateStringFromCallReply(RedisModuleCallReply *reply) {
  *     "!" -> REDISMODULE_ARGV_REPLICATE
  *     "A" -> REDISMODULE_ARGV_NO_AOF
  *     "R" -> REDISMODULE_ARGV_NO_REPLICAS
+ *     "3" -> REDISMODULE_ARGV_RESP_3
+ *     "0" -> REDISMODULE_ARGV_RESP_AUTO
  *
  * On error (format specifier error) NULL is returned and nothing is
  * allocated. On success the argument vector is returned. */
@@ -4198,6 +4144,10 @@ robj **moduleCreateArgvFromUserFormat(const char *cmdname, const char *fmt, int
             if (flags) (*flags) |= REDISMODULE_ARGV_NO_AOF;
         } else if (*p == 'R') {
             if (flags) (*flags) |= REDISMODULE_ARGV_NO_REPLICAS;
+        } else if (*p == '3') {
+            if (flags) (*flags) |= REDISMODULE_ARGV_RESP_3;
+        } else if (*p == '0') {
+            if (flags) (*flags) |= REDISMODULE_ARGV_RESP_AUTO;
         } else {
             goto fmterr;
         }
@@ -4218,7 +4168,7 @@ fmterr:
  * * **cmdname**: The Redis command to call.
  * * **fmt**: A format specifier string for the command's arguments. Each
  *   of the arguments should be specified by a valid type specification. The
- *   format specifier can also contain the modifiers `!`, `A` and `R` which
+ *   format specifier can also contain the modifiers `!`, `A`, `3` and `R` which
  *   don't have a corresponding argument.
  *
  *     * `b` -- The argument is a buffer and is immediately followed by another
@@ -4230,6 +4180,11 @@ fmterr:
  *     * `!` -- Sends the Redis command and its arguments to replicas and AOF.
  *     * `A` -- Suppress AOF propagation, send only to replicas (requires `!`).
  *     * `R` -- Suppress replicas propagation, send only to AOF (requires `!`).
+ *     * `3` -- Return a RESP3 reply. This will change the command reply.
+ *              e.g., HGETALL returns a map instead of a flat array.
+ *     * `0` -- Return the reply in auto mode, i.e. the reply format will be the
+ *              same as the client attached to the given RedisModuleCtx. This will
+ *              probably used when you want to pass the reply directly to the client.
  * * **...**: The actual arguments to the Redis command.
  *
  * On success a RedisModuleCallReply object is returned, otherwise
@@ -4288,6 +4243,13 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
     c->db = ctx->client->db;
     c->argv = argv;
     c->argc = argc;
+    c->resp = 2;
+    if (flags & REDISMODULE_ARGV_RESP_3) {
+        c->resp = 3;
+    } else if (flags & REDISMODULE_ARGV_RESP_AUTO) {
+        /* Auto mode means to take the same protocol as the ctx client. */
+        c->resp = ctx->client->resp;
+    }
     if (ctx->module) ctx->module->in_call++;
 
     /* We handle the above format error only when the client is setup so that
@@ -4374,7 +4336,7 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
         proto = sdscatlen(proto,o->buf,o->used);
         listDelNode(c->reply,listFirst(c->reply));
     }
-    reply = moduleCreateCallReplyFromProto(ctx,proto);
+    reply = callReplyCreate(proto, ctx);
     autoMemoryAdd(ctx,REDISMODULE_AM_REPLY,reply);
 
 cleanup:
@@ -4397,8 +4359,7 @@ cleanup:
 /* Return a pointer, and a length, to the protocol returned by the command
  * that returned the reply object. */
 const char *RM_CallReplyProto(RedisModuleCallReply *reply, size_t *len) {
-    if (reply->proto) *len = sdslen(reply->proto);
-    return reply->proto;
+    return callReplyGetProto(reply, len);
 }
 
 /* --------------------------------------------------------------------------
@@ -9592,6 +9553,14 @@ void moduleRegisterCoreAPI(void) {
     REGISTER_API(CallReplyProto);
     REGISTER_API(FreeCallReply);
     REGISTER_API(CallReplyInteger);
+    REGISTER_API(CallReplyDouble);
+    REGISTER_API(CallReplyBigNumber);
+    REGISTER_API(CallReplyVerbatim);
+    REGISTER_API(CallReplyBool);
+    REGISTER_API(CallReplySetElement);
+    REGISTER_API(CallReplyMapElement);
+    REGISTER_API(CallReplyAttributeElement);
+    REGISTER_API(CallReplyAttribute);
     REGISTER_API(CallReplyType);
     REGISTER_API(CallReplyLength);
     REGISTER_API(CallReplyArrayElement);
diff --git a/src/redismodule.h b/src/redismodule.h
index df2e854fe..66543ffbc 100644
--- a/src/redismodule.h
+++ b/src/redismodule.h
@@ -46,6 +46,13 @@
 #define REDISMODULE_REPLY_INTEGER 2
 #define REDISMODULE_REPLY_ARRAY 3
 #define REDISMODULE_REPLY_NULL 4
+#define REDISMODULE_REPLY_MAP 5
+#define REDISMODULE_REPLY_SET 6
+#define REDISMODULE_REPLY_BOOL 7
+#define REDISMODULE_REPLY_DOUBLE 8
+#define REDISMODULE_REPLY_BIG_NUMBER 9
+#define REDISMODULE_REPLY_VERBATIM_STRING 10
+#define REDISMODULE_REPLY_ATTRIBUTE 11
 
 /* Postponed array length. */
 #define REDISMODULE_POSTPONED_ARRAY_LEN -1  /* Deprecated, please use REDISMODULE_POSTPONED_LEN */
@@ -139,11 +146,13 @@ typedef struct RedisModuleStreamID {
 /* The current client does not allow blocking, either called from
  * within multi, lua, or from another module using RM_Call */
 #define REDISMODULE_CTX_FLAGS_DENY_BLOCKING (1<<21)
+/* The current client uses RESP3 protocol */
+#define REDISMODULE_CTX_FLAGS_RESP3 (1<<22)
 
 /* Next context flag, must be updated when adding new flags above!
 This flag should not be used directly by the module.
  * Use RedisModule_GetContextFlagsAll instead. */
-#define _REDISMODULE_CTX_FLAGS_NEXT (1<<22)
+#define _REDISMODULE_CTX_FLAGS_NEXT (1<<23)
 
 /* Keyspace changes notification classes. Every class is associated with a
  * character for configuration purposes.
@@ -610,6 +619,14 @@ REDISMODULE_API const char * (*RedisModule_CallReplyProto)(RedisModuleCallReply
 REDISMODULE_API void (*RedisModule_FreeCallReply)(RedisModuleCallReply *reply) REDISMODULE_ATTR;
 REDISMODULE_API int (*RedisModule_CallReplyType)(RedisModuleCallReply *reply) REDISMODULE_ATTR;
 REDISMODULE_API long long (*RedisModule_CallReplyInteger)(RedisModuleCallReply *reply) REDISMODULE_ATTR;
+REDISMODULE_API double (*RedisModule_CallReplyDouble)(RedisModuleCallReply *reply) REDISMODULE_ATTR;
+REDISMODULE_API int (*RedisModule_CallReplyBool)(RedisModuleCallReply *reply) REDISMODULE_ATTR;
+REDISMODULE_API const char* (*RedisModule_CallReplyBigNumber)(RedisModuleCallReply *reply, size_t *len) REDISMODULE_ATTR;
+REDISMODULE_API const char* (*RedisModule_CallReplyVerbatim)(RedisModuleCallReply *reply, size_t *len, const char **format) REDISMODULE_ATTR;
+REDISMODULE_API RedisModuleCallReply * (*RedisModule_CallReplySetElement)(RedisModuleCallReply *reply, size_t idx) REDISMODULE_ATTR;
+REDISMODULE_API int (*RedisModule_CallReplyMapElement)(RedisModuleCallReply *reply, size_t idx, RedisModuleCallReply **key, RedisModuleCallReply **val) REDISMODULE_ATTR;
+REDISMODULE_API int (*RedisModule_CallReplyAttributeElement)(RedisModuleCallReply *reply, size_t idx, RedisModuleCallReply **key, RedisModuleCallReply **val) REDISMODULE_ATTR;
+REDISMODULE_API RedisModuleCallReply * (*RedisModule_CallReplyAttribute)(RedisModuleCallReply *reply) REDISMODULE_ATTR;
 REDISMODULE_API size_t (*RedisModule_CallReplyLength)(RedisModuleCallReply *reply) REDISMODULE_ATTR;
 REDISMODULE_API RedisModuleCallReply * (*RedisModule_CallReplyArrayElement)(RedisModuleCallReply *reply, size_t idx) REDISMODULE_ATTR;
 REDISMODULE_API RedisModuleString * (*RedisModule_CreateString)(RedisModuleCtx *ctx, const char *ptr, size_t len) REDISMODULE_ATTR;
@@ -931,6 +948,14 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
     REDISMODULE_GET_API(CallReplyProto);
     REDISMODULE_GET_API(FreeCallReply);
     REDISMODULE_GET_API(CallReplyInteger);
+    REDISMODULE_GET_API(CallReplyDouble);
+    REDISMODULE_GET_API(CallReplyBool);
+    REDISMODULE_GET_API(CallReplyBigNumber);
+    REDISMODULE_GET_API(CallReplyVerbatim);
+    REDISMODULE_GET_API(CallReplySetElement);
+    REDISMODULE_GET_API(CallReplyMapElement);
+    REDISMODULE_GET_API(CallReplyAttributeElement);
+    REDISMODULE_GET_API(CallReplyAttribute);
     REDISMODULE_GET_API(CallReplyType);
     REDISMODULE_GET_API(CallReplyLength);
     REDISMODULE_GET_API(CallReplyArrayElement);
diff --git a/src/resp_parser.c b/src/resp_parser.c
new file mode 100644
index 000000000..c9f5d0b00
--- /dev/null
+++ b/src/resp_parser.c
@@ -0,0 +1,228 @@
+/*
+ * Copyright (c) 2009-2021, Redis Labs Ltd.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *   * Redistributions of source code must retain the above copyright notice,
+ *     this list of conditions and the following disclaimer.
+ *   * Redistributions in binary form must reproduce the above copyright
+ *     notice, this list of conditions and the following disclaimer in the
+ *     documentation and/or other materials provided with the distribution.
+ *   * Neither the name of Redis nor the names of its contributors may be used
+ *     to endorse or promote products derived from this software without
+ *     specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+/* ----------------------------------------------------------------------------------------
+ * A RESP parser for parsing replies returned by RM_Call or Lua's
+ * 'redis.call()'.
+ *
+ * The parser introduces callbacks that need to be set by the user. Each
+ * callback represents a different reply type. Each callback gets a p_ctx that
+ * was given to the parseReply function. The callbacks also give the protocol
+ * (underlying blob) of the current reply and the size.
+ *
+ * Some callbacks also get the parser object itself:
+ * - array_callback
+ * - set_callback
+ * - map_callback
+ * 
+ * These callbacks need to continue parsing by calling parseReply a number of
+ * times, according to the supplied length. Subsequent parseReply calls may use
+ * a different p_ctx, which will be used for nested CallReply objects.
+ *
+ * These callbacks also do not receive a proto_len, which is not known at the
+ * time of parsing. Callers may calculate it themselves after parsing the
+ * entire collection.
+ *
+ * NOTE: This parser is designed to only handle replies generated by Redis
+ * itself. It does not perform many required validations and thus NOT SAFE FOR
+ * PARSING USER INPUT.
+ * ----------------------------------------------------------------------------------------
+ */
+
+#include "resp_parser.h"
+#include "server.h"
+
+static int parseBulk(ReplyParser *parser, void *p_ctx) {
+    const char *proto = parser->curr_location;
+    char *p = strchr(proto+1,'\r');
+    long long bulklen;
+    parser->curr_location = p + 2; /* for \r\n */
+
+    string2ll(proto+1,p-proto-1,&bulklen);
+    if (bulklen == -1) {
+        parser->callbacks.null_bulk_string_callback(p_ctx, proto, parser->curr_location - proto);
+    } else {
+        const char *str = parser->curr_location;
+        parser->curr_location += bulklen;
+        parser->curr_location += 2; /* for \r\n */
+        parser->callbacks.bulk_string_callback(p_ctx, str, bulklen, proto, parser->curr_location - proto);
+    }
+
+    return C_OK;
+}
+
+static int parseSimpleString(ReplyParser *parser, void *p_ctx) {
+    const char *proto = parser->curr_location;
+    char *p = strchr(proto+1,'\r');
+    parser->curr_location = p + 2; /* for \r\n */
+    parser->callbacks.simple_str_callback(p_ctx, proto+1, p-proto-1, proto, parser->curr_location - proto);
+    return C_OK;
+}
+
+static int parseError(ReplyParser *parser, void *p_ctx) {
+    const char *proto = parser->curr_location;
+    char *p = strchr(proto+1,'\r');
+    parser->curr_location = p + 2; // for \r\n
+    parser->callbacks.error_callback(p_ctx, proto+1, p-proto-1, proto, parser->curr_location - proto);
+    return C_OK;
+}
+
+static int parseLong(ReplyParser *parser, void *p_ctx) {
+    const char *proto = parser->curr_location;
+    char *p = strchr(proto+1,'\r');
+    parser->curr_location = p + 2; /* for \r\n */
+    long long val;
+    string2ll(proto+1,p-proto-1,&val);
+    parser->callbacks.long_callback(p_ctx, val, proto, parser->curr_location - proto);
+    return C_OK;
+}
+
+static int parseAttributes(ReplyParser *parser, void *p_ctx) {
+    const char *proto = parser->curr_location;
+    char *p = strchr(proto+1,'\r');
+    long long len;
+    string2ll(proto+1,p-proto-1,&len);
+    p += 2;
+    parser->curr_location = p;
+    parser->callbacks.attribute_callback(parser, p_ctx, len, proto);
+    return C_OK;
+}
+
+static int parseVerbatimString(ReplyParser *parser, void *p_ctx) {
+    const char *proto = parser->curr_location;
+    char *p = strchr(proto+1,'\r');
+    long long bulklen;
+    parser->curr_location = p + 2; /* for \r\n */
+    string2ll(proto+1,p-proto-1,&bulklen);
+    const char *format = parser->curr_location;
+    parser->curr_location += bulklen;
+    parser->curr_location += 2; /* for \r\n */
+    parser->callbacks.verbatim_string_callback(p_ctx, format, format + 4, bulklen - 4, proto, parser->curr_location - proto);
+    return C_OK;
+}
+
+static int parseBigNumber(ReplyParser *parser, void *p_ctx) {
+    const char *proto = parser->curr_location;
+    char *p = strchr(proto+1,'\r');
+    parser->curr_location = p + 2; /* for \r\n */
+    parser->callbacks.big_number_callback(p_ctx, proto+1, p-proto-1, proto, parser->curr_location - proto);
+    return C_OK;
+}
+
+static int parseNull(ReplyParser *parser, void *p_ctx) {
+    const char *proto = parser->curr_location;
+    char *p = strchr(proto+1,'\r');
+    parser->curr_location = p + 2; /* for \r\n */
+    parser->callbacks.null_callback(p_ctx, proto, parser->curr_location - proto);
+    return C_OK;
+}
+
+static int parseDouble(ReplyParser *parser, void *p_ctx) {
+    const char *proto = parser->curr_location;
+    char *p = strchr(proto+1,'\r');
+    parser->curr_location = p + 2; /* for \r\n */
+    char buf[MAX_LONG_DOUBLE_CHARS+1];
+    size_t len = p-proto-1;
+    double d;
+    if (len <= MAX_LONG_DOUBLE_CHARS) {
+        memcpy(buf,proto+1,len);
+        buf[len] = '\0';
+        d = strtod(buf,NULL); /* We expect a valid representation. */
+    } else {
+        d = 0;
+    }
+    parser->callbacks.double_callback(p_ctx, d, proto, parser->curr_location - proto);
+    return C_OK;
+}
+
+static int parseBool(ReplyParser *parser, void *p_ctx) {
+    const char *proto = parser->curr_location;
+    char *p = strchr(proto+1,'\r');
+    parser->curr_location = p + 2; /* for \r\n */
+    parser->callbacks.bool_callback(p_ctx, proto[1] == 't', proto, parser->curr_location - proto);
+    return C_OK;
+}
+
+static int parseArray(ReplyParser *parser, void *p_ctx) {
+    const char *proto = parser->curr_location;
+    char *p = strchr(proto+1,'\r');
+    long long len;
+    string2ll(proto+1,p-proto-1,&len);
+    p += 2;
+    parser->curr_location = p;
+    if (len == -1) {
+        parser->callbacks.null_array_callback(p_ctx, proto, parser->curr_location - proto);
+    } else {
+        parser->callbacks.array_callback(parser, p_ctx, len, proto);
+    }
+    return C_OK;
+}
+
+static int parseSet(ReplyParser *parser, void *p_ctx) {
+    const char *proto = parser->curr_location;
+    char *p = strchr(proto+1,'\r');
+    long long len;
+    string2ll(proto+1,p-proto-1,&len);
+    p += 2;
+    parser->curr_location = p;
+    parser->callbacks.set_callback(parser, p_ctx, len, proto);
+    return C_OK;
+}
+
+static int parseMap(ReplyParser *parser, void *p_ctx) {
+    const char *proto = parser->curr_location;
+    char *p = strchr(proto+1,'\r');
+    long long len;
+    string2ll(proto+1,p-proto-1,&len);
+    p += 2;
+    parser->curr_location = p;
+    parser->callbacks.map_callback(parser, p_ctx, len, proto);
+    return C_OK;
+}
+
+/* Parse a reply pointed to by parser->curr_location. */
+int parseReply(ReplyParser *parser, void *p_ctx) {
+    switch (parser->curr_location[0]) {
+        case '$': return parseBulk(parser, p_ctx);
+        case '+': return parseSimpleString(parser, p_ctx);
+        case '-': return parseError(parser, p_ctx);
+        case ':': return parseLong(parser, p_ctx);
+        case '*': return parseArray(parser, p_ctx);
+        case '~': return parseSet(parser, p_ctx);
+        case '%': return parseMap(parser, p_ctx);
+        case '#': return parseBool(parser, p_ctx);
+        case ',': return parseDouble(parser, p_ctx);
+        case '_': return parseNull(parser, p_ctx);
+        case '(': return parseBigNumber(parser, p_ctx);
+        case '=': return parseVerbatimString(parser, p_ctx);
+        case '|': return parseAttributes(parser, p_ctx);
+        default: if (parser->callbacks.error) parser->callbacks.error(p_ctx);
+    }
+    return C_ERR;
+}
diff --git a/src/resp_parser.h b/src/resp_parser.h
new file mode 100644
index 000000000..4597efee3
--- /dev/null
+++ b/src/resp_parser.h
@@ -0,0 +1,94 @@
+/*
+ * Copyright (c) 2021, Redis Labs Ltd.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *   * Redistributions of source code must retain the above copyright notice,
+ *     this list of conditions and the following disclaimer.
+ *   * Redistributions in binary form must reproduce the above copyright
+ *     notice, this list of conditions and the following disclaimer in the
+ *     documentation and/or other materials provided with the distribution.
+ *   * Neither the name of Redis nor the names of its contributors may be used
+ *     to endorse or promote products derived from this software without
+ *     specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#ifndef SRC_RESP_PARSER_H_
+#define SRC_RESP_PARSER_H_
+
+#include <stddef.h>
+
+typedef struct ReplyParser ReplyParser;
+
+typedef struct ReplyParserCallbacks {
+    /* Called when the parser reaches an empty mbulk ('*-1') */
+    void (*null_array_callback)(void *ctx, const char *proto, size_t proto_len);
+
+    /* Called when the parser reaches an empty bulk ('$-1') (bulk len is -1) */
+    void (*null_bulk_string_callback)(void *ctx, const char *proto, size_t proto_len);
+
+    /* Called when the parser reaches a bulk ('$'), which is passed as 'str' along with its length 'len' */
+    void (*bulk_string_callback)(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len);
+
+    /* Called when the parser reaches an error ('-'), which is passed as 'str' along with its length 'len' */
+    void (*error_callback)(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len);
+
+    /* Called when the parser reaches a simple string ('+'), which is passed as 'str' along with its length 'len' */
+    void (*simple_str_callback)(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len);
+
+    /* Called when the parser reaches a long long value (':'), which is passed as an argument 'val' */
+    void (*long_callback)(void *ctx, long long val, const char *proto, size_t proto_len);
+
+    /* Called when the parser reaches an array ('*'). The array length is passed as an argument 'len' */
+    void (*array_callback)(struct ReplyParser *parser, void *ctx, size_t len, const char *proto);
+
+    /* Called when the parser reaches a set ('~'). The set length is passed as an argument 'len' */
+    void (*set_callback)(struct ReplyParser *parser, void *ctx, size_t len, const char *proto);
+
+    /* Called when the parser reaches a map ('%'). The map length is passed as an argument 'len' */
+    void (*map_callback)(struct ReplyParser *parser, void *ctx, size_t len, const char *proto);
+
+    /* Called when the parser reaches a bool ('#'), which is passed as an argument 'val' */
+    void (*bool_callback)(void *ctx, int val, const char *proto, size_t proto_len);
+
+    /* Called when the parser reaches a double (','), which is passed as an argument 'val' */
+    void (*double_callback)(void *ctx, double val, const char *proto, size_t proto_len);
+
+    /* Called when the parser reaches a big number (','), which is passed as 'str' along with its length 'len' */
+    void (*big_number_callback)(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len);
+
+    /* Called when the parser reaches a string, which is passed as 'str' along with its 'format' and length 'len' */
+    void (*verbatim_string_callback)(void *ctx, const char *format, const char *str, size_t len, const char *proto, size_t proto_len);
+
+    /* Called when the parser reaches an attribute ('|'). The attribute length is passed as an argument 'len' */
+    void (*attribute_callback)(struct ReplyParser *parser, void *ctx, size_t len, const char *proto);
+
+    /* Called when the parser reaches a null ('_') */
+    void (*null_callback)(void *ctx, const char *proto, size_t proto_len);
+
+    void (*error)(void *ctx);
+} ReplyParserCallbacks;
+
+struct ReplyParser {
+    /* The current location in the reply buffer, needs to be set to the beginning of the reply */
+    const char *curr_location;
+    ReplyParserCallbacks callbacks;
+};
+
+int parseReply(ReplyParser *parser, void *p_ctx);
+
+#endif /* SRC_RESP_PARSER_H_ */
diff --git a/src/scripting.c b/src/scripting.c
index b485d01ed..87b3b036b 100644
--- a/src/scripting.c
+++ b/src/scripting.c
@@ -32,6 +32,7 @@
 #include "rand.h"
 #include "cluster.h"
 #include "monotonic.h"
+#include "resp_parser.h"
 
 #include <lua.h>
 #include <lauxlib.h>
@@ -39,14 +40,21 @@
 #include <ctype.h>
 #include <math.h>
 
-char *redisProtocolToLuaType_Int(lua_State *lua, char *reply);
-char *redisProtocolToLuaType_Bulk(lua_State *lua, char *reply);
-char *redisProtocolToLuaType_Status(lua_State *lua, char *reply);
-char *redisProtocolToLuaType_Error(lua_State *lua, char *reply);
-char *redisProtocolToLuaType_Aggregate(lua_State *lua, char *reply, int atype);
-char *redisProtocolToLuaType_Null(lua_State *lua, char *reply);
-char *redisProtocolToLuaType_Bool(lua_State *lua, char *reply, int tf);
-char *redisProtocolToLuaType_Double(lua_State *lua, char *reply);
+static void redisProtocolToLuaType_Int(void *ctx, long long val, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_BulkString(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_NullBulkString(void *ctx, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_NullArray(void *ctx, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_Status(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_Error(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_Array(struct ReplyParser *parser, void *ctx, size_t len, const char *proto);
+static void redisProtocolToLuaType_Map(struct ReplyParser *parser, void *ctx, size_t len, const char *proto);
+static void redisProtocolToLuaType_Set(struct ReplyParser *parser, void *ctx, size_t len, const char *proto);
+static void redisProtocolToLuaType_Null(void *ctx, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_Bool(void *ctx, int val, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_Double(void *ctx, double d, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_BigNumber(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_VerbatimString(void *ctx, const char *format, const char *str, size_t len, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_Attribute(struct ReplyParser *parser, void *ctx, size_t len, const char *proto);
 int redis_math_random (lua_State *L);
 int redis_math_randomseed (lua_State *L);
 void ldbInit(void);
@@ -128,139 +136,238 @@ void sha1hex(char *digest, char *script, size_t len) {
  * error string.
  */
 
-char *redisProtocolToLuaType(lua_State *lua, char* reply) {
-    char *p = reply;
-
-    switch(*p) {
-    case ':': p = redisProtocolToLuaType_Int(lua,reply); break;
-    case '$': p = redisProtocolToLuaType_Bulk(lua,reply); break;
-    case '+': p = redisProtocolToLuaType_Status(lua,reply); break;
-    case '-': p = redisProtocolToLuaType_Error(lua,reply); break;
-    case '*': p = redisProtocolToLuaType_Aggregate(lua,reply,*p); break;
-    case '%': p = redisProtocolToLuaType_Aggregate(lua,reply,*p); break;
-    case '~': p = redisProtocolToLuaType_Aggregate(lua,reply,*p); break;
-    case '_': p = redisProtocolToLuaType_Null(lua,reply); break;
-    case '#': p = redisProtocolToLuaType_Bool(lua,reply,p[1]); break;
-    case ',': p = redisProtocolToLuaType_Double(lua,reply); break;
+static const ReplyParserCallbacks DefaultLuaTypeParserCallbacks = {
+    .null_array_callback = redisProtocolToLuaType_NullArray,
+    .bulk_string_callback = redisProtocolToLuaType_BulkString,
+    .null_bulk_string_callback = redisProtocolToLuaType_NullBulkString,
+    .error_callback = redisProtocolToLuaType_Error,
+    .simple_str_callback = redisProtocolToLuaType_Status,
+    .long_callback = redisProtocolToLuaType_Int,
+    .array_callback = redisProtocolToLuaType_Array,
+    .set_callback = redisProtocolToLuaType_Set,
+    .map_callback = redisProtocolToLuaType_Map,
+    .bool_callback = redisProtocolToLuaType_Bool,
+    .double_callback = redisProtocolToLuaType_Double,
+    .null_callback = redisProtocolToLuaType_Null,
+    .big_number_callback = redisProtocolToLuaType_BigNumber,
+    .verbatim_string_callback = redisProtocolToLuaType_VerbatimString,
+    .attribute_callback = redisProtocolToLuaType_Attribute,
+    .error = NULL,
+};
+
+void redisProtocolToLuaType(lua_State *lua, char* reply) {
+    ReplyParser parser = {.curr_location = reply, .callbacks = DefaultLuaTypeParserCallbacks};
+
+    parseReply(&parser, lua);
+}
+
+static void redisProtocolToLuaType_Int(void *ctx, long long val, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
     }
-    return p;
+
+    lua_State *lua = ctx;
+    lua_pushnumber(lua,(lua_Number)val);
 }
 
-char *redisProtocolToLuaType_Int(lua_State *lua, char *reply) {
-    char *p = strchr(reply+1,'\r');
-    long long value;
+static void redisProtocolToLuaType_NullBulkString(void *ctx, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
+    }
 
-    string2ll(reply+1,p-reply-1,&value);
-    lua_pushnumber(lua,(lua_Number)value);
-    return p+2;
+    lua_State *lua = ctx;
+    lua_pushboolean(lua,0);
 }
 
-char *redisProtocolToLuaType_Bulk(lua_State *lua, char *reply) {
-    char *p = strchr(reply+1,'\r');
-    long long bulklen;
+static void redisProtocolToLuaType_NullArray(void *ctx, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
+    }
+    lua_State *lua = ctx;
+    lua_pushboolean(lua,0);
+}
 
-    string2ll(reply+1,p-reply-1,&bulklen);
-    if (bulklen == -1) {
-        lua_pushboolean(lua,0);
-        return p+2;
-    } else {
-        lua_pushlstring(lua,p+2,bulklen);
-        return p+2+bulklen+2;
+
+static void redisProtocolToLuaType_BulkString(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
     }
+
+    lua_State *lua = ctx;
+    lua_pushlstring(lua,str,len);
 }
 
-char *redisProtocolToLuaType_Status(lua_State *lua, char *reply) {
-    char *p = strchr(reply+1,'\r');
+static void redisProtocolToLuaType_Status(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
+    }
+
+    lua_State *lua = ctx;
 
     lua_newtable(lua);
     lua_pushstring(lua,"ok");
-    lua_pushlstring(lua,reply+1,p-reply-1);
+    lua_pushlstring(lua,str,len);
     lua_settable(lua,-3);
-    return p+2;
 }
 
-char *redisProtocolToLuaType_Error(lua_State *lua, char *reply) {
-    char *p = strchr(reply+1,'\r');
+static void redisProtocolToLuaType_Error(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
+    }
+
+    lua_State *lua = ctx;
 
     lua_newtable(lua);
     lua_pushstring(lua,"err");
-    lua_pushlstring(lua,reply+1,p-reply-1);
+    lua_pushlstring(lua,str,len);
     lua_settable(lua,-3);
-    return p+2;
 }
 
-char *redisProtocolToLuaType_Aggregate(lua_State *lua, char *reply, int atype) {
-    char *p = strchr(reply+1,'\r');
-    long long mbulklen;
-    int j = 0;
-
-    string2ll(reply+1,p-reply-1,&mbulklen);
-    if (server.lua_client->resp == 2 || atype == '*') {
-        p += 2;
-        if (mbulklen == -1) {
-            lua_pushboolean(lua,0);
-            return p;
-        }
+static void redisProtocolToLuaType_Map(struct ReplyParser *parser, void *ctx, size_t len, const char *proto) {
+    UNUSED(proto);
+    lua_State *lua = ctx;
+    if (lua) {
         lua_newtable(lua);
-        for (j = 0; j < mbulklen; j++) {
-            lua_pushnumber(lua,j+1);
-            p = redisProtocolToLuaType(lua,p);
-            lua_settable(lua,-3);
-        }
-    } else if (server.lua_client->resp == 3) {
-        /* Here we handle only Set and Map replies in RESP3 mode, since arrays
-         * follow the above RESP2 code path. Note that those are represented
-         * as a table with the "map" or "set" field populated with the actual
-         * table representing the set or the map type. */
-        p += 2;
+        lua_pushstring(lua, "map");
+        lua_newtable(lua);
+    }
+    for (size_t j = 0; j < len; j++) {
+        parseReply(parser,lua);
+        parseReply(parser,lua);
+        if (lua) lua_settable(lua,-3);
+    }
+    if (lua) lua_settable(lua,-3);
+}
+
+static void redisProtocolToLuaType_Set(struct ReplyParser *parser, void *ctx, size_t len, const char *proto) {
+    UNUSED(proto);
+
+    lua_State *lua = ctx;
+    if (lua) {
         lua_newtable(lua);
-        lua_pushstring(lua,atype == '%' ? "map" : "set");
+        lua_pushstring(lua, "set");
         lua_newtable(lua);
-        for (j = 0; j < mbulklen; j++) {
-            p = redisProtocolToLuaType(lua,p);
-            if (atype == '%') {
-                p = redisProtocolToLuaType(lua,p);
-            } else {
-                lua_pushboolean(lua,1);
-            }
+    }
+    for (size_t j = 0; j < len; j++) {
+        parseReply(parser,lua);
+        if (lua) {
+            lua_pushboolean(lua,1);
             lua_settable(lua,-3);
         }
-        lua_settable(lua,-3);
     }
-    return p;
+    if (lua) lua_settable(lua,-3);
 }
 
-char *redisProtocolToLuaType_Null(lua_State *lua, char *reply) {
-    char *p = strchr(reply+1,'\r');
+static void redisProtocolToLuaType_Array(struct ReplyParser *parser, void *ctx, size_t len, const char *proto) {
+    UNUSED(proto);
+
+    lua_State *lua = ctx;
+    if (lua) lua_newtable(lua);
+    for (size_t j = 0; j < len; j++) {
+        if (lua) lua_pushnumber(lua,j+1);
+        parseReply(parser,lua);
+        if (lua) lua_settable(lua,-3);
+    }
+}
+
+static void redisProtocolToLuaType_Attribute(struct ReplyParser *parser, void *ctx, size_t len, const char *proto) {
+    UNUSED(proto);
+
+    /* Parse the attribute reply.
+     * Currently, we do not expose the attribute to the Lua script so
+     * we just need to continue parsing and ignore it (the NULL ensures that the
+     * reply will be ignored). */
+    for (size_t j = 0; j < len; j++) {
+        parseReply(parser,NULL);
+        parseReply(parser,NULL);
+    }
+
+    /* Parse the reply itself. */
+    parseReply(parser,ctx);
+}
+
+static void redisProtocolToLuaType_VerbatimString(void *ctx, const char *format, const char *str, size_t len, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
+    }
+
+    lua_State *lua = ctx;
+
+    lua_newtable(lua);
+    lua_pushstring(lua,"verbatim_string");
+    lua_newtable(lua);
+    lua_pushstring(lua,"string");
+    lua_pushlstring(lua,str,len);
+    lua_settable(lua,-3);
+    lua_pushstring(lua,"format");
+    lua_pushlstring(lua,format,3);
+    lua_settable(lua,-3);
+    lua_settable(lua,-3);
+}
+
+static void redisProtocolToLuaType_BigNumber(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
+    }
+
+    lua_State *lua = ctx;
+
+    lua_newtable(lua);
+    lua_pushstring(lua,"big_number");
+    lua_pushlstring(lua,str,len);
+    lua_settable(lua,-3);
+}
+
+static void redisProtocolToLuaType_Null(void *ctx, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
+    }
+
+    lua_State *lua = ctx;
     lua_pushnil(lua);
-    return p+2;
 }
 
-char *redisProtocolToLuaType_Bool(lua_State *lua, char *reply, int tf) {
-    char *p = strchr(reply+1,'\r');
-    lua_pushboolean(lua,tf == 't');
-    return p+2;
+static void redisProtocolToLuaType_Bool(void *ctx, int val, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
+    }
+
+    lua_State *lua = ctx;
+    lua_pushboolean(lua,val);
 }
 
-char *redisProtocolToLuaType_Double(lua_State *lua, char *reply) {
-    char *p = strchr(reply+1,'\r');
-    char buf[MAX_LONG_DOUBLE_CHARS+1];
-    size_t len = p-reply-1;
-    double d;
-
-    if (len <= MAX_LONG_DOUBLE_CHARS) {
-        memcpy(buf,reply+1,len);
-        buf[len] = '\0';
-        d = strtod(buf,NULL); /* We expect a valid representation. */
-    } else {
-        d = 0;
+static void redisProtocolToLuaType_Double(void *ctx, double d, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
     }
 
+    lua_State *lua = ctx;
     lua_newtable(lua);
     lua_pushstring(lua,"double");
     lua_pushnumber(lua,d);
     lua_settable(lua,-3);
-    return p+2;
 }
 
 /* This function is used in order to push an error on the Lua stack in the
@@ -398,6 +505,43 @@ void luaReplyToRedisReply(client *c, lua_State *lua) {
         }
         lua_pop(lua,1); /* Discard field name pushed before. */
 
+        /* Handle big number reply. */
+        lua_pushstring(lua,"big_number");
+        lua_gettable(lua,-2);
+        t = lua_type(lua,-1);
+        if (t == LUA_TSTRING) {
+            addReplyBigNum(c,(char*)lua_tostring(lua,-1),lua_strlen(lua,-1));
+            lua_pop(lua,2);
+            return;
+        }
+        lua_pop(lua,1); /* Discard field name pushed before. */
+
+        /* Handle verbatim reply. */
+        lua_pushstring(lua,"verbatim_string");
+        lua_gettable(lua,-2);
+        t = lua_type(lua,-1);
+        if (t == LUA_TTABLE) {
+            lua_pushstring(lua,"format");
+            lua_gettable(lua,-2);
+            t = lua_type(lua,-1);
+            if (t == LUA_TSTRING){
+                char* format = (char*)lua_tostring(lua,-1);
+                lua_pushstring(lua,"string");
+                lua_gettable(lua,-3);
+                t = lua_type(lua,-1);
+                if (t == LUA_TSTRING){
+                    size_t len;
+                    char* str = (char*)lua_tolstring(lua,-1,&len);
+                    addReplyVerbatim(c, str, len, format);
+                    lua_pop(lua,4);
+                    return;
+                }
+                lua_pop(lua,1);
+            }
+            lua_pop(lua,1);
+        }
+        lua_pop(lua,1); /* Discard field name pushed before. */
+
         /* Handle map reply. */
         lua_pushstring(lua,"map");
         lua_gettable(lua,-2);
@@ -598,7 +742,7 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) {
     c->cmd = c->lastcmd = cmd;
 
     /* There are commands that are not allowed inside scripts. */
-    if (cmd->flags & CMD_NOSCRIPT) {
+    if (!server.lua_disable_deny_script && (cmd->flags & CMD_NOSCRIPT)) {
         luaPushError(lua, "This Redis command is not allowed from scripts");
         goto cleanup;
     }
@@ -1115,6 +1259,7 @@ void scriptingInit(int setup) {
         server.lua_caller = NULL;
         server.lua_cur_script = NULL;
         server.lua_timedout = 0;
+        server.lua_disable_deny_script = 0;
         ldbInit();
     }
 
diff --git a/src/server.h b/src/server.h
index 72e8a0103..f9572276f 100644
--- a/src/server.h
+++ b/src/server.h
@@ -1635,6 +1635,7 @@ struct redisServer {
     int lua_kill;         /* Kill the script if true. */
     int lua_always_replicate_commands; /* Default replication type. */
     int lua_oom;          /* OOM detected when script start? */
+    int lua_disable_deny_script; /* Allow running commands marked "no-script" inside a script. */
     /* Lazy free */
     int lazyfree_lazy_eviction;
     int lazyfree_lazy_expire;
@@ -1890,6 +1891,7 @@ void addReplyErrorSds(client *c, sds err);
 void addReplyError(client *c, const char *err);
 void addReplyStatus(client *c, const char *status);
 void addReplyDouble(client *c, double d);
+void addReplyLongLongWithPrefix(client *c, long long ll, char prefix);
 void addReplyBigNum(client *c, const char* num, size_t len);
 void addReplyHumanLongDouble(client *c, long double d);
 void addReplyLongLong(client *c, long long ll);