Repository: redis/redis-py
Pull request #2269 (2022-07-10 20:14:48)

redis stream example
### Pull Request check-list

_Please make sure to review and check all of these items:_

- [ ] Does `$ tox` pass with this change (including linting)?
- [ ] Do the CI tests pass with this change (enable it first in your forked repo and wait for the github action build to finish)?
- [ ] Is the new or changed code fully tested?
- [ ] Is a documentation update included (if this change modifies existing APIs, or introduces new ones)?
- [ ] Is there an example added to the examples folder (if applicable)?
- [ ] Was the change added to CHANGES file?

_NOTE: these things are not required to open a PR and can be done
afterwards / while the PR is open._

### Description of change

Redis streams examples only.


codecov-commenter: # [Codecov](https://codecov.io/gh/redis/redis-py/pull/2269?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) Report
> Merging [#2269](https://codecov.io/gh/redis/redis-py/pull/2269?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) (1d57f4b) into [master](https://codecov.io/gh/redis/redis-py/commit/4b0543d567aef36ac467ce495d831a24575d8d5b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) (4b0543d) will **decrease** coverage by `0.02%`.
> The diff coverage is `n/a`.

```diff
@@            Coverage Diff             @@
##           master    #2269      +/-   ##
==========================================
- Coverage   91.87%   91.85%   -0.03%     
==========================================
  Files         108      108              
  Lines       27783    27782       -1     
==========================================
- Hits        25527    25518       -9     
- Misses       2256     2264       +8     
```


| [Impacted Files](https://codecov.io/gh/redis/redis-py/pull/2269?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) | Coverage Δ | |
|---|---|---|
| [tests/conftest.py](https://codecov.io/gh/redis/redis-py/pull/2269/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvY29uZnRlc3QucHk=) | `86.50% <0.00%> (-0.80%)` | :arrow_down: |
| [tests/test\_asyncio/test\_search.py](https://codecov.io/gh/redis/redis-py/pull/2269/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3Rfc2VhcmNoLnB5) | `98.35% <0.00%> (-0.33%)` | :arrow_down: |
| [redis/asyncio/connection.py](https://codecov.io/gh/redis/redis-py/pull/2269/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvYXN5bmNpby9jb25uZWN0aW9uLnB5) | `83.84% <0.00%> (-0.23%)` | :arrow_down: |
| [tests/test\_asyncio/test\_cluster.py](https://codecov.io/gh/redis/redis-py/pull/2269/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfY2x1c3Rlci5weQ==) | `97.44% <0.00%> (-0.14%)` | :arrow_down: |
| [tests/test\_graph.py](https://codecov.io/gh/redis/redis-py/pull/2269/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9ncmFwaC5weQ==) | `90.65% <0.00%> (-0.04%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/redis/redis-py/pull/2269?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/redis/redis-py/pull/2269?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis). Last update [4b0543d...1d57f4b](https://codecov.io/gh/redis/redis-py/pull/2269?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).

pedrofrazao: > Can you add the files to docs/examples.rst? Also why did you split your examples into 2 different files?

Yes, I can add to docs/examples.rst.
The split is a logical one. Simple stream usage and stream groups usage.
But it's more logical to merge, I can merge it.

Patch:
diff --git a/CHANGES b/CHANGES
index 0af421b4..6a890ef1 100644
--- a/CHANGES
+++ b/CHANGES
@@ -1,4 +1,5 @@
-
+    * Remove verbose logging from cluster.py
+    * Add retry mechanism to async version of Connection
     * Compare commands case-insensitively in the asyncio command parser
     * Allow negative `retries` for `Retry` class to retry forever
     * Add `items` parameter to `hset` signature
diff --git a/docs/examples.rst b/docs/examples.rst
index 722fae2d..08526ff6 100644
--- a/docs/examples.rst
+++ b/docs/examples.rst
@@ -12,3 +12,4 @@ Examples
    examples/set_and_get_examples
    examples/search_vector_similarity_examples
    examples/pipeline_examples
+   examples/redis-stream-example.ipynb
diff --git a/docs/examples/redis-stream-example.ipynb b/docs/examples/redis-stream-example.ipynb
new file mode 100644
index 00000000..9303b527
--- /dev/null
+++ b/docs/examples/redis-stream-example.ipynb
@@ -0,0 +1,754 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# Redis Stream Examples"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## basic config"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 1,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "redis_host = \"redis\"\n",
+    "stream_key = \"skey\"\n",
+    "stream2_key = \"s2key\"\n",
+    "group1 = \"grp1\"\n",
+    "group2 = \"grp2\""
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## connection"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 2,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "True"
+      ]
+     },
+     "execution_count": 2,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "import redis\n",
+    "from time import time\n",
+    "from redis.exceptions import ConnectionError, DataError, NoScriptError, RedisError, ResponseError\n",
+    "\n",
+    "r = redis.Redis( redis_host )\n",
+    "r.ping()"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## xadd and xread"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### add some data to the stream"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 3,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "stream length: 10\n"
+     ]
+    }
+   ],
+   "source": [
+    "for i in range(0,10):\n",
+    "    r.xadd( stream_key, { 'ts': time(), 'v': i } )\n",
+    "print( f\"stream length: {r.xlen( stream_key )}\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### read some data from the stream"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 4,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "[[b'skey', [(b'1657571033115-0', {b'ts': b'1657571033.1128936', b'v': b'0'}), (b'1657571033117-0', {b'ts': b'1657571033.1176307', b'v': b'1'})]]]\n"
+     ]
+    }
+   ],
+   "source": [
+    "## read 2 entries from stream_key\n",
+    "l = r.xread( count=2, streams={stream_key:0} )\n",
+    "print(l)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### extract data from the returned structure"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 5,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "got data from stream: b'skey'\n",
+      "id: b'1657571033115-0' value: b'0'\n",
+      "id: b'1657571033117-0' value: b'1'\n"
+     ]
+    }
+   ],
+   "source": [
+    "first_stream = l[0]\n",
+    "print( f\"got data from stream: {first_stream[0]}\")\n",
+    "fs_data = first_stream[1]\n",
+    "for id, value in fs_data:\n",
+    "    print( f\"id: {id} value: {value[b'v']}\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### read more data from the stream\n",
+    "if we call the `xread` with the same arguments we will get the same data"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 6,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "id: b'1657571033115-0' value: b'0'\n",
+      "id: b'1657571033117-0' value: b'1'\n"
+     ]
+    }
+   ],
+   "source": [
+    "l = r.xread( count=2, streams={stream_key:0} )\n",
+    "for id, value in l[0][1]:\n",
+    "    print( f\"id: {id} value: {value[b'v']}\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "to get new data we need to change the key passed to the call"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 7,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "id: b'1657571033118-0' value: b'2'\n",
+      "id: b'1657571033119-0' value: b'3'\n"
+     ]
+    }
+   ],
+   "source": [
+    "last_id_returned = l[0][1][-1][0]\n",
+    "l = r.xread( count=2, streams={stream_key: last_id_returned} )\n",
+    "for id, value in l[0][1]:\n",
+    "    print( f\"id: {id} value: {value[b'v']}\")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 8,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "id: b'1657571033119-1' value: b'4'\n",
+      "id: b'1657571033121-0' value: b'5'\n"
+     ]
+    }
+   ],
+   "source": [
+    "last_id_returned = l[0][1][-1][0]\n",
+    "l = r.xread( count=2, streams={stream_key: last_id_returned} )\n",
+    "for id, value in l[0][1]:\n",
+    "    print( f\"id: {id} value: {value[b'v']}\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "to get only newer entries"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 9,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "stream length: 10\n",
+      "after 5s block, got an empty list [], no *new* messages on the stream\n",
+      "stream length: 10\n"
+     ]
+    }
+   ],
+   "source": [
+    "print( f\"stream length: {r.xlen( stream_key )}\")\n",
+    "# wait for 5s for new messages\n",
+    "l = r.xread( count=1, block=5000, streams={stream_key: '$'} )\n",
+    "print( f\"after 5s block, got an empty list {l}, no *new* messages on the stream\")\n",
+    "print( f\"stream length: {r.xlen( stream_key )}\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### 2nd stream\n",
+    "Add some messages to a 2nd stream"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 10,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "stream length: 10\n"
+     ]
+    }
+   ],
+   "source": [
+    "for i in range(1000,1010):\n",
+    "    r.xadd( stream2_key, { 'v': i } )\n",
+    "print( f\"stream length: {r.xlen( stream2_key )}\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "get messages from the 2 streams"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 11,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "got from b'skey' the entry [(b'1657571033115-0', {b'ts': b'1657571033.1128936', b'v': b'0'})]\n",
+      "got from b's2key' the entry [(b'1657571042111-0', {b'v': b'1000'})]\n"
+     ]
+    }
+   ],
+   "source": [
+    "l = r.xread( count=1, streams={stream_key:0,stream2_key:0} )\n",
+    "for k,d in l:\n",
+    "    print(f\"got from {k} the entry {d}\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# stream groups\n",
+    "With the groups is possible track, for many consumers, and at the Redis side, which message have been already consumed.\n",
+    "## add some data to streams\n",
+    "Creating 2 streams with 10 messages each."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 12,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "stream 'skey' length: 20\n",
+      "stream 's2key' length: 20\n"
+     ]
+    }
+   ],
+   "source": [
+    "def add_some_data_to_stream( sname, key_range ):\n",
+    "    for i in key_range:\n",
+    "        r.xadd( sname, { 'ts': time(), 'v': i } )\n",
+    "    print( f\"stream '{sname}' length: {r.xlen( stream_key )}\")\n",
+    "\n",
+    "add_some_data_to_stream( stream_key, range(0,10) )\n",
+    "add_some_data_to_stream( stream2_key, range(1000,1010) )"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## use a group to read from the stream\n",
+    "* create a group `grp1` with the stream `skey`, and\n",
+    "* create a group `grp2` with the streams `skey` and `s2key`\n",
+    "\n",
+    "Use the `xinfo_group` to verify the result of the group creation."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 13,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "skey -> group name: b'grp1' with 0 consumers and b'0-0' as last read id\n",
+      "skey -> group name: b'grp2' with 0 consumers and b'0-0' as last read id\n",
+      "s2key -> group name: b'grp2' with 0 consumers and b'0-0' as last read id\n"
+     ]
+    }
+   ],
+   "source": [
+    "## create the group\n",
+    "def create_group( skey, gname ):\n",
+    "    try:\n",
+    "        r.xgroup_create( name=skey, groupname=gname, id=0 )\n",
+    "    except ResponseError as e:\n",
+    "        print(f\"raised: {e}\")\n",
+    "\n",
+    "# group1 read the stream 'skey'\n",
+    "create_group( stream_key, group1 )\n",
+    "# group2 read the streams 'skey' and 's2key'\n",
+    "create_group( stream_key, group2 )\n",
+    "create_group( stream2_key, group2 )\n",
+    "\n",
+    "def group_info( skey ):\n",
+    "    res = r.xinfo_groups( name=skey )\n",
+    "    for i in res:\n",
+    "        print( f\"{skey} -> group name: {i['name']} with {i['consumers']} consumers and {i['last-delivered-id']}\"\n",
+    "              + f\" as last read id\")\n",
+    "    \n",
+    "group_info( stream_key )\n",
+    "group_info( stream2_key )"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## group read\n",
+    "The `xreadgroup` method permit to read from a stream group."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 14,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "def print_xreadgroup_reply( reply, group = None, run = None):\n",
+    "    for d_stream in reply:\n",
+    "        for element in d_stream[1]:\n",
+    "            print(  f\"got element {element[0]}\"\n",
+    "                  + f\"from stream {d_stream[0]}\" )\n",
+    "            if run is not None:\n",
+    "                run( d_stream[0], group, element[0] )"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 15,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "got element b'1657571033115-0'from stream b'skey'\n",
+      "got element b'1657571033117-0'from stream b'skey'\n"
+     ]
+    }
+   ],
+   "source": [
+    "# read some messages on group1 with consumer 'c' \n",
+    "d = r.xreadgroup( groupname=group1, consumername='c', block=10,\n",
+    "                  count=2, streams={stream_key:'>'})\n",
+    "print_xreadgroup_reply( d )"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "A **2nd consumer** for the same stream group will get not delivered messages."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 16,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "got element b'1657571033118-0'from stream b'skey'\n",
+      "got element b'1657571033119-0'from stream b'skey'\n"
+     ]
+    }
+   ],
+   "source": [
+    "# read some messages on group1 with consumer 'c' \n",
+    "d = r.xreadgroup( groupname=group1, consumername='c2', block=10,\n",
+    "                  count=2, streams={stream_key:'>'})\n",
+    "print_xreadgroup_reply( d )"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "But a **2nd stream group** can read the already delivered messages again.\n",
+    "\n",
+    "Note that the 2nd stream group include also the 2nd stream.\n",
+    "That can be identified in the reply (1st element of the reply list)."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 18,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "got element b'1657571033115-0'from stream b'skey'\n",
+      "got element b'1657571033117-0'from stream b'skey'\n",
+      "got element b'1657571042111-0'from stream b's2key'\n",
+      "got element b'1657571042113-0'from stream b's2key'\n"
+     ]
+    }
+   ],
+   "source": [
+    "d2 = r.xreadgroup( groupname=group2, consumername='c', block=10,\n",
+    "                   count=2, streams={stream_key:'>',stream2_key:'>'})\n",
+    "print_xreadgroup_reply( d2 )"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "To check for pending messages (delivered messages without acknowledgment) we can use the `xpending`."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 19,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "4 pending messages on 'skey' for group 'grp1'\n",
+      "2 pending messages on 'skey' for group 'grp2'\n",
+      "2 pending messages on 's2key' for group 'grp2'\n"
+     ]
+    }
+   ],
+   "source": [
+    "# check pending status (read messages without a ack)\n",
+    "def print_pending_info( key_group ):\n",
+    "    for s,k in key_group:\n",
+    "        pr = r.xpending( name=s, groupname=k )\n",
+    "        print( f\"{pr.get('pending')} pending messages on '{s}' for group '{k}'\" )\n",
+    "    \n",
+    "print_pending_info( ((stream_key,group1),(stream_key,group2),(stream2_key,group2)) )"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## ack\n",
+    "Acknowledge some messages with `xack`."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 20,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "got element b'1657571033118-0'from stream b'skey'\n",
+      "got element b'1657571033119-0'from stream b'skey'\n"
+     ]
+    }
+   ],
+   "source": [
+    "# do acknowledges for group1\n",
+    "toack = lambda k,g,e: r.xack( k,g, e )\n",
+    "print_xreadgroup_reply( d, group=group1, run=toack )"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 21,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "2 pending messages on 'skey' for group 'grp1'\n",
+      "2 pending messages on 'skey' for group 'grp2'\n",
+      "2 pending messages on 's2key' for group 'grp2'\n"
+     ]
+    }
+   ],
+   "source": [
+    "# check pending again\n",
+    "print_pending_info( ((stream_key,group1),(stream_key,group2),(stream2_key,group2)) )"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "ack all messages on the `group1`."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 22,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "got element b'1657571033119-1'from stream b'skey'\n",
+      "got element b'1657571033121-0'from stream b'skey'\n",
+      "got element b'1657571033121-1'from stream b'skey'\n",
+      "got element b'1657571033121-2'from stream b'skey'\n",
+      "got element b'1657571033122-0'from stream b'skey'\n",
+      "got element b'1657571033122-1'from stream b'skey'\n",
+      "got element b'1657571049557-0'from stream b'skey'\n",
+      "got element b'1657571049557-1'from stream b'skey'\n",
+      "got element b'1657571049558-0'from stream b'skey'\n",
+      "got element b'1657571049559-0'from stream b'skey'\n",
+      "got element b'1657571049559-1'from stream b'skey'\n",
+      "got element b'1657571049559-2'from stream b'skey'\n",
+      "got element b'1657571049560-0'from stream b'skey'\n",
+      "got element b'1657571049562-0'from stream b'skey'\n",
+      "got element b'1657571049563-0'from stream b'skey'\n",
+      "got element b'1657571049563-1'from stream b'skey'\n",
+      "2 pending messages on 'skey' for group 'grp1'\n"
+     ]
+    }
+   ],
+   "source": [
+    "d = r.xreadgroup( groupname=group1, consumername='c', block=10,\n",
+    "                      count=100, streams={stream_key:'>'})\n",
+    "print_xreadgroup_reply( d, group=group1, run=toack)\n",
+    "print_pending_info( ((stream_key,group1),) )"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "But stream length will be the same after the `xack` of all messages on the `group1`."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 24,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "20"
+      ]
+     },
+     "execution_count": 24,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "r.xlen(stream_key)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## delete all\n",
+    "To remove the messages with need to remote them explicitly with `xdel`."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 25,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "s1 = r.xread( streams={stream_key:0} )\n",
+    "for streams in s1:\n",
+    "    stream_name, messages = streams\n",
+    "    # del all ids from the message list\n",
+    "    [ r.xdel( stream_name, i[0] ) for i in messages ]"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "stream length"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 26,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "0"
+      ]
+     },
+     "execution_count": 26,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "r.xlen(stream_key)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "But with the `xdel` the 2nd group can read any not processed message from the `skey`."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 27,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "got element b'1657571042113-1'from stream b's2key'\n",
+      "got element b'1657571042114-0'from stream b's2key'\n"
+     ]
+    }
+   ],
+   "source": [
+    "d2 = r.xreadgroup( groupname=group2, consumername='c', block=10,\n",
+    "                   count=2, streams={stream_key:'>',stream2_key:'>'})\n",
+    "print_xreadgroup_reply( d2 )"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": []
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.8.10"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 4
+}
diff --git a/redis/asyncio/cluster.py b/redis/asyncio/cluster.py
index 28940044..3fe3ebc4 100644
--- a/redis/asyncio/cluster.py
+++ b/redis/asyncio/cluster.py
@@ -755,7 +755,6 @@ class ClusterNode:
     """
 
     __slots__ = (
-        "_command_stack",
         "_connections",
         "_free",
         "connection_class",
@@ -796,7 +795,6 @@ class ClusterNode:
 
         self._connections: List[Connection] = []
         self._free: Deque[Connection] = collections.deque(maxlen=self.max_connections)
-        self._command_stack: List["PipelineCommand"] = []
 
     def __repr__(self) -> str:
         return (
@@ -887,18 +885,18 @@ class ClusterNode:
             # Release connection
             self._free.append(connection)
 
-    async def execute_pipeline(self) -> bool:
+    async def execute_pipeline(self, commands: List["PipelineCommand"]) -> bool:
         # Acquire connection
         connection = self.acquire_connection()
 
         # Execute command
         await connection.send_packed_command(
-            connection.pack_commands(cmd.args for cmd in self._command_stack), False
+            connection.pack_commands(cmd.args for cmd in commands), False
         )
 
         # Read responses
         ret = False
-        for cmd in self._command_stack:
+        for cmd in commands:
             try:
                 cmd.result = await self.parse_response(
                     connection, cmd.args[0], **cmd.kwargs
@@ -1365,12 +1363,14 @@ class ClusterPipeline(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterComm
 
             node = target_nodes[0]
             if node.name not in nodes:
-                nodes[node.name] = node
-                node._command_stack = []
-            node._command_stack.append(cmd)
+                nodes[node.name] = (node, [])
+            nodes[node.name][1].append(cmd)
 
         errors = await asyncio.gather(
-            *(asyncio.ensure_future(node.execute_pipeline()) for node in nodes.values())
+            *(
+                asyncio.ensure_future(node[0].execute_pipeline(node[1]))
+                for node in nodes.values()
+            )
         )
 
         if any(errors):
diff --git a/redis/asyncio/connection.py b/redis/asyncio/connection.py
index 35536fc8..4c75d2f1 100644
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -637,6 +637,8 @@ class Connection:
             retry_on_error = []
         if retry_on_timeout:
             retry_on_error.append(TimeoutError)
+            retry_on_error.append(socket.timeout)
+            retry_on_error.append(asyncio.TimeoutError)
         self.retry_on_error = retry_on_error
         if retry_on_error:
             if not retry:
@@ -685,7 +687,7 @@ class Connection:
 
     @property
     def is_connected(self):
-        return self._reader and self._writer
+        return self._reader is not None and self._writer is not None
 
     def register_connect_callback(self, callback):
         self._connect_callbacks.append(weakref.WeakMethod(callback))
@@ -706,7 +708,9 @@ class Connection:
         if self.is_connected:
             return
         try:
-            await self._connect()
+            await self.retry.call_with_retry(
+                lambda: self._connect(), lambda error: self.disconnect()
+            )
         except asyncio.CancelledError:
             raise
         except (socket.timeout, asyncio.TimeoutError):
@@ -767,7 +771,16 @@ class Connection:
     def _error_message(self, exception):
         # args for socket.error can either be (errno, "message")
         # or just "message"
-        if len(exception.args) == 1:
+        if not exception.args:
+            # asyncio has a bug where on Connection reset by peer, the
+            # exception is not instanciated, so args is empty. This is the
+            # workaround.
+            # See: https://github.com/redis/redis-py/issues/2237
+            # See: https://github.com/python/cpython/issues/94061
+            return (
+                f"Error connecting to {self.host}:{self.port}. Connection reset by peer"
+            )
+        elif len(exception.args) == 1:
             return f"Error connecting to {self.host}:{self.port}. {exception.args[0]}."
         else:
             return (
diff --git a/redis/asyncio/sentinel.py b/redis/asyncio/sentinel.py
index 5aefd09e..99c50749 100644
--- a/redis/asyncio/sentinel.py
+++ b/redis/asyncio/sentinel.py
@@ -44,7 +44,7 @@ class SentinelManagedConnection(Connection):
             if str_if_bytes(await self.read_response()) != "PONG":
                 raise ConnectionError("PING failed")
 
-    async def connect(self):
+    async def _connect_retry(self):
         if self._reader:
             return  # already connected
         if self.connection_pool.is_master:
@@ -57,6 +57,12 @@ class SentinelManagedConnection(Connection):
                     continue
             raise SlaveNotFoundError  # Never be here
 
+    async def connect(self):
+        return await self.retry.call_with_retry(
+            self._connect_retry,
+            lambda error: asyncio.sleep(0),
+        )
+
     async def read_response(self, disable_decoding: bool = False):
         try:
             return await super().read_response(disable_decoding=disable_decoding)
diff --git a/redis/cluster.py b/redis/cluster.py
index 6034e960..b2d4f3b0 100644
--- a/redis/cluster.py
+++ b/redis/cluster.py
@@ -1,5 +1,4 @@
 import copy
-import logging
 import random
 import socket
 import sys
@@ -15,7 +14,6 @@ from redis.crc import REDIS_CLUSTER_HASH_SLOTS, key_slot
 from redis.exceptions import (
     AskError,
     AuthenticationError,
-    BusyLoadingError,
     ClusterCrossSlotError,
     ClusterDownError,
     ClusterError,
@@ -39,8 +37,6 @@ from redis.utils import (
     str_if_bytes,
 )
 
-log = logging.getLogger(__name__)
-
 
 def get_node_name(host: str, port: int) -> str:
     return f"{host}:{port}"
@@ -535,7 +531,6 @@ class RedisCluster(AbstractRedisCluster, RedisClusterCommands):
                 " RedisCluster(startup_nodes=[ClusterNode('localhost', 6379),"
                 " ClusterNode('localhost', 6378)])"
             )
-        log.debug(f"startup_nodes : {startup_nodes}")
         # Update the connection arguments
         # Whenever a new connection is established, RedisCluster's on_connect
         # method should be run
@@ -666,13 +661,8 @@ class RedisCluster(AbstractRedisCluster, RedisClusterCommands):
         :return True if the default node was set, else False
         """
         if node is None or self.get_node(node_name=node.name) is None:
-            log.info(
-                "The requested node does not exist in the cluster, so "
-                "the default node was not changed."
-            )
             return False
         self.nodes_manager.default_node = node
-        log.info(f"Changed the default cluster node to {node}")
         return True
 
     def monitor(self, target_node=None):
@@ -816,8 +806,6 @@ class RedisCluster(AbstractRedisCluster, RedisClusterCommands):
         else:
             # get the nodes group for this command if it was predefined
             command_flag = self.command_flags.get(command)
-        if command_flag:
-            log.debug(f"Target node/s for {command}: {command_flag}")
         if command_flag == self.__class__.RANDOM:
             # return a random node
             return [self.get_random_node()]
@@ -841,7 +829,6 @@ class RedisCluster(AbstractRedisCluster, RedisClusterCommands):
             node = self.nodes_manager.get_node_from_slot(
                 slot, self.read_from_replicas and command in READ_COMMANDS
             )
-            log.debug(f"Target for {args}: slot {slot}")
             return [node]
 
     def _should_reinitialized(self):
@@ -1019,7 +1006,7 @@ class RedisCluster(AbstractRedisCluster, RedisClusterCommands):
                     res[node.name] = self._execute_command(node, *args, **kwargs)
                 # Return the processed result
                 return self._process_result(args[0], res, **kwargs)
-            except BaseException as e:
+            except Exception as e:
                 if type(e) in self.__class__.ERRORS_ALLOW_RETRY:
                     # The nodes and slots cache were reinitialized.
                     # Try again with the new cluster setup.
@@ -1059,10 +1046,6 @@ class RedisCluster(AbstractRedisCluster, RedisClusterCommands):
                     )
                     moved = False
 
-                log.debug(
-                    f"Executing command {command} on target node: "
-                    f"{target_node.server_type} {target_node.name}"
-                )
                 redis_node = self.get_redis_connection(target_node)
                 connection = get_connection(redis_node, *args, **kwargs)
                 if asking:
@@ -1077,12 +1060,9 @@ class RedisCluster(AbstractRedisCluster, RedisClusterCommands):
                         response, **kwargs
                     )
                 return response
-
-            except (RedisClusterException, BusyLoadingError, AuthenticationError) as e:
-                log.exception(type(e))
+            except AuthenticationError:
                 raise
             except (ConnectionError, TimeoutError) as e:
-                log.exception(type(e))
                 # ConnectionError can also be raised if we couldn't get a
                 # connection from the pool before timing out, so check that
                 # this is an actual connection before attempting to disconnect.
@@ -1101,7 +1081,7 @@ class RedisCluster(AbstractRedisCluster, RedisClusterCommands):
                     # and try again with the new setup
                     target_node.redis_connection = None
                     self.nodes_manager.initialize()
-                    raise
+                    raise e
             except MovedError as e:
                 # First, we will try to patch the slots/nodes cache with the
                 # redirected node output and try again. If MovedError exceeds
@@ -1111,7 +1091,6 @@ class RedisCluster(AbstractRedisCluster, RedisClusterCommands):
                 # the same client object is shared between multiple threads. To
                 # reduce the frequency you can set this variable in the
                 # RedisCluster constructor.
-                log.exception("MovedError")
                 self.reinitialize_counter += 1
                 if self._should_reinitialized():
                     self.nodes_manager.initialize()
@@ -1121,29 +1100,21 @@ class RedisCluster(AbstractRedisCluster, RedisClusterCommands):
                     self.nodes_manager.update_moved_exception(e)
                 moved = True
             except TryAgainError:
-                log.exception("TryAgainError")
-
                 if ttl < self.RedisClusterRequestTTL / 2:
                     time.sleep(0.05)
             except AskError as e:
-                log.exception("AskError")
-
                 redirect_addr = get_node_name(host=e.host, port=e.port)
                 asking = True
             except ClusterDownError as e:
-                log.exception("ClusterDownError")
                 # ClusterDownError can occur during a failover and to get
                 # self-healed, we will try to reinitialize the cluster layout
                 # and retry executing the command
                 time.sleep(0.25)
                 self.nodes_manager.initialize()
                 raise e
-            except ResponseError as e:
-                message = e.__str__()
-                log.exception(f"ResponseError: {message}")
-                raise e
-            except BaseException as e:
-                log.exception("BaseException")
+            except ResponseError:
+                raise
+            except Exception as e:
                 if connection:
                     connection.disconnect()
                 raise e
@@ -1280,11 +1251,6 @@ class NodesManager:
         elif node_name:
             return self.nodes_cache.get(node_name)
         else:
-            log.error(
-                "get_node requires one of the following: "
-                "1. node name "
-                "2. host and port"
-            )
             return None
 
     def update_moved_exception(self, exception):
@@ -1432,7 +1398,6 @@ class NodesManager:
         :startup_nodes:
             Responsible for discovering other nodes in the cluster
         """
-        log.debug("Initializing the nodes' topology of the cluster")
         self.reset()
         tmp_nodes_cache = {}
         tmp_slots = {}
@@ -1460,17 +1425,9 @@ class NodesManager:
                     )
                 cluster_slots = str_if_bytes(r.execute_command("CLUSTER SLOTS"))
                 startup_nodes_reachable = True
-            except (ConnectionError, TimeoutError) as e:
-                msg = e.__str__
-                log.exception(
-                    "An exception occurred while trying to"
-                    " initialize the cluster using the seed node"
-                    f" {startup_node.name}:\n{msg}"
-                )
+            except (ConnectionError, TimeoutError):
                 continue
             except ResponseError as e:
-                log.exception('ReseponseError sending "cluster slots" to redis server')
-
                 # Isn't a cluster connection, so it won't parse these
                 # exceptions automatically
                 message = e.__str__()
@@ -2025,7 +1982,7 @@ class ClusterPipeline(RedisCluster):
         )
         if attempt and allow_redirections:
             # RETRY MAGIC HAPPENS HERE!
-            # send these remaing comamnds one at a time using `execute_command`
+            # send these remaing commands one at a time using `execute_command`
             # in the main client. This keeps our retry logic
             # in one place mostly,
             # and allows us to be more confident in correctness of behavior.
@@ -2042,12 +1999,6 @@ class ClusterPipeline(RedisCluster):
             # If a lot of commands have failed, we'll be setting the
             # flag to rebuild the slots table from scratch.
             # So MOVED errors should correct themselves fairly quickly.
-            log.exception(
-                f"An exception occurred during pipeline execution. "
-                f"args: {attempt[-1].args}, "
-                f"error: {type(attempt[-1].result).__name__} "
-                f"{str(attempt[-1].result)}"
-            )
             self.reinitialize_counter += 1
             if self._should_reinitialized():
                 self.nodes_manager.initialize()
diff --git a/redis/commands/core.py b/redis/commands/core.py
index 6d67415a..027d3dbc 100644
--- a/redis/commands/core.py
+++ b/redis/commands/core.py
@@ -5459,7 +5459,7 @@ class GeoCommands(CommandsProtocol):
         `m` for meters (the default value), `km` for kilometers,
         `mi` for miles and `ft` for feet.
         ``sort`` indicates to return the places in a sorted way,
-        ASC for nearest to farest and DESC for farest to nearest.
+        ASC for nearest to furthest and DESC for furthest to nearest.
         ``count`` limit the results to the first count matching items.
         ``any`` is set to True, the command will return as soon as
         enough matches are found. Can't be provided without ``count``
diff --git a/redis/commands/search/result.py b/redis/commands/search/result.py
index 5f4aca64..451bf89b 100644
--- a/redis/commands/search/result.py
+++ b/redis/commands/search/result.py
@@ -38,7 +38,7 @@ class Result:
             score = float(res[i + 1]) if with_scores else None
 
             fields = {}
-            if hascontent:
+            if hascontent and res[i + fields_offset] is not None:
                 fields = (
                     dict(
                         dict(
diff --git a/redis/commands/timeseries/commands.py b/redis/commands/timeseries/commands.py
index 3a30c246..13e3cdf4 100644
--- a/redis/commands/timeseries/commands.py
+++ b/redis/commands/timeseries/commands.py
@@ -1,4 +1,7 @@
+from typing import Dict, List, Optional, Tuple, Union
+
 from redis.exceptions import DataError
+from redis.typing import KeyT, Number
 
 ADD_CMD = "TS.ADD"
 ALTER_CMD = "TS.ALTER"
@@ -22,7 +25,15 @@ REVRANGE_CMD = "TS.REVRANGE"
 class TimeSeriesCommands:
     """RedisTimeSeries Commands."""
 
-    def create(self, key, **kwargs):
+    def create(
+        self,
+        key: KeyT,
+        retention_msecs: Optional[int] = None,
+        uncompressed: Optional[bool] = False,
+        labels: Optional[Dict[str, str]] = None,
+        chunk_size: Optional[int] = None,
+        duplicate_policy: Optional[str] = None,
+    ):
         """
         Create a new time-series.
 
@@ -31,40 +42,26 @@ class TimeSeriesCommands:
         key:
             time-series key
         retention_msecs:
-            Maximum age for samples compared to last event time (in milliseconds).
+            Maximum age for samples compared to highest reported timestamp (in milliseconds).
             If None or 0 is passed then  the series is not trimmed at all.
         uncompressed:
-            Since RedisTimeSeries v1.2, both timestamps and values are
-            compressed by default.
-            Adding this flag will keep data in an uncompressed form.
-            Compression not only saves
-            memory but usually improve performance due to lower number
-            of memory accesses.
+            Changes data storage from compressed (by default) to uncompressed
         labels:
             Set of label-value pairs that represent metadata labels of the key.
         chunk_size:
-            Each time-serie uses chunks of memory of fixed size for
-            time series samples.
-            You can alter the default TSDB chunk size by passing the
-            chunk_size argument (in Bytes).
+            Memory size, in bytes, allocated for each data chunk.
+            Must be a multiple of 8 in the range [128 .. 1048576].
         duplicate_policy:
-            Since RedisTimeSeries v1.4 you can specify the duplicate sample policy
-            ( Configure what to do on duplicate sample. )
+            Policy for handling multiple samples with identical timestamps.
             Can be one of:
             - 'block': an error will occur for any out of order sample.
             - 'first': ignore the new value.
             - 'last': override with latest value.
             - 'min': only override if the value is lower than the existing value.
             - 'max': only override if the value is higher than the existing value.
-            When this is not set, the server-wide default will be used.
 
-        For more information: https://oss.redis.com/redistimeseries/commands/#tscreate
+        For more information: https://redis.io/commands/ts.create/
         """  # noqa
-        retention_msecs = kwargs.get("retention_msecs", None)
-        uncompressed = kwargs.get("uncompressed", False)
-        labels = kwargs.get("labels", {})
-        chunk_size = kwargs.get("chunk_size", None)
-        duplicate_policy = kwargs.get("duplicate_policy", None)
         params = [key]
         self._append_retention(params, retention_msecs)
         self._append_uncompressed(params, uncompressed)
@@ -74,29 +71,62 @@ class TimeSeriesCommands:
 
         return self.execute_command(CREATE_CMD, *params)
 
-    def alter(self, key, **kwargs):
+    def alter(
+        self,
+        key: KeyT,
+        retention_msecs: Optional[int] = None,
+        labels: Optional[Dict[str, str]] = None,
+        chunk_size: Optional[int] = None,
+        duplicate_policy: Optional[str] = None,
+    ):
         """
-        Update the retention, labels of an existing key.
-        For more information see
+        Update the retention, chunk size, duplicate policy, and labels of an existing
+        time series.
+
+        Args:
 
-        The parameters are the same as TS.CREATE.
+        key:
+            time-series key
+        retention_msecs:
+            Maximum retention period, compared to maximal existing timestamp (in milliseconds).
+            If None or 0 is passed then  the series is not trimmed at all.
+        labels:
+            Set of label-value pairs that represent metadata labels of the key.
+        chunk_size:
+            Memory size, in bytes, allocated for each data chunk.
+            Must be a multiple of 8 in the range [128 .. 1048576].
+        duplicate_policy:
+            Policy for handling multiple samples with identical timestamps.
+            Can be one of:
+            - 'block': an error will occur for any out of order sample.
+            - 'first': ignore the new value.
+            - 'last': override with latest value.
+            - 'min': only override if the value is lower than the existing value.
+            - 'max': only override if the value is higher than the existing value.
 
-        For more information: https://oss.redis.com/redistimeseries/commands/#tsalter
+        For more information: https://redis.io/commands/ts.alter/
         """  # noqa
-        retention_msecs = kwargs.get("retention_msecs", None)
-        labels = kwargs.get("labels", {})
-        duplicate_policy = kwargs.get("duplicate_policy", None)
         params = [key]
         self._append_retention(params, retention_msecs)
+        self._append_chunk_size(params, chunk_size)
         self._append_duplicate_policy(params, ALTER_CMD, duplicate_policy)
         self._append_labels(params, labels)
 
         return self.execute_command(ALTER_CMD, *params)
 
-    def add(self, key, timestamp, value, **kwargs):
+    def add(
+        self,
+        key: KeyT,
+        timestamp: Union[int, str],
+        value: Number,
+        retention_msecs: Optional[int] = None,
+        uncompressed: Optional[bool] = False,
+        labels: Optional[Dict[str, str]] = None,
+        chunk_size: Optional[int] = None,
+        duplicate_policy: Optional[str] = None,
+    ):
         """
-        Append (or create and append) a new sample to the series.
-        For more information see
+        Append (or create and append) a new sample to a time series.
 
         Args:
 
@@ -107,35 +137,26 @@ class TimeSeriesCommands:
         value:
             Numeric data value of the sample
         retention_msecs:
-            Maximum age for samples compared to last event time (in milliseconds).
+            Maximum retention period, compared to maximal existing timestamp (in milliseconds).
             If None or 0 is passed then  the series is not trimmed at all.
         uncompressed:
-            Since RedisTimeSeries v1.2, both timestamps and values are compressed by default.
-            Adding this flag will keep data in an uncompressed form. Compression not only saves
-            memory but usually improve performance due to lower number of memory accesses.
+            Changes data storage from compressed (by default) to uncompressed
         labels:
             Set of label-value pairs that represent metadata labels of the key.
         chunk_size:
-            Each time-serie uses chunks of memory of fixed size for time series samples.
-            You can alter the default TSDB chunk size by passing the chunk_size argument (in Bytes).
+            Memory size, in bytes, allocated for each data chunk.
+            Must be a multiple of 8 in the range [128 .. 1048576].
         duplicate_policy:
-            Since RedisTimeSeries v1.4 you can specify the duplicate sample policy
-            (Configure what to do on duplicate sample).
+            Policy for handling multiple samples with identical timestamps.
             Can be one of:
             - 'block': an error will occur for any out of order sample.
             - 'first': ignore the new value.
             - 'last': override with latest value.
             - 'min': only override if the value is lower than the existing value.
             - 'max': only override if the value is higher than the existing value.
-            When this is not set, the server-wide default will be used.
 
-        For more information: https://oss.redis.com/redistimeseries/master/commands/#tsadd
+        For more information: https://redis.io/commands/ts.add/
         """  # noqa
-        retention_msecs = kwargs.get("retention_msecs", None)
-        uncompressed = kwargs.get("uncompressed", False)
-        labels = kwargs.get("labels", {})
-        chunk_size = kwargs.get("chunk_size", None)
-        duplicate_policy = kwargs.get("duplicate_policy", None)
         params = [key, timestamp, value]
         self._append_retention(params, retention_msecs)
         self._append_uncompressed(params, uncompressed)
@@ -145,28 +166,34 @@ class TimeSeriesCommands:
 
         return self.execute_command(ADD_CMD, *params)
 
-    def madd(self, ktv_tuples):
+    def madd(self, ktv_tuples: List[Tuple[KeyT, Union[int, str], Number]]):
         """
         Append (or create and append) a new `value` to series
         `key` with `timestamp`.
         Expects a list of `tuples` as (`key`,`timestamp`, `value`).
         Return value is an array with timestamps of insertions.
 
-        For more information: https://oss.redis.com/redistimeseries/master/commands/#tsmadd
+        For more information: https://redis.io/commands/ts.madd/
         """  # noqa
         params = []
         for ktv in ktv_tuples:
-            for item in ktv:
-                params.append(item)
+            params.extend(ktv)
 
         return self.execute_command(MADD_CMD, *params)
 
-    def incrby(self, key, value, **kwargs):
+    def incrby(
+        self,
+        key: KeyT,
+        value: Number,
+        timestamp: Optional[Union[int, str]] = None,
+        retention_msecs: Optional[int] = None,
+        uncompressed: Optional[bool] = False,
+        labels: Optional[Dict[str, str]] = None,
+        chunk_size: Optional[int] = None,
+    ):
         """
-        Increment (or create an time-series and increment) the latest
-        sample's of a series.
-        This command can be used as a counter or gauge that automatically gets
-        history as a time series.
+        Increment (or create an time-series and increment) the latest sample's of a series.
+        This command can be used as a counter or gauge that automatically gets history as a time series.
 
         Args:
 
@@ -175,27 +202,19 @@ class TimeSeriesCommands:
         value:
             Numeric data value of the sample
         timestamp:
-            Timestamp of the sample. None can be used for automatic timestamp (using the system clock).
+            Timestamp of the sample. * can be used for automatic timestamp (using the system clock).
         retention_msecs:
             Maximum age for samples compared to last event time (in milliseconds).
             If None or 0 is passed then  the series is not trimmed at all.
         uncompressed:
-            Since RedisTimeSeries v1.2, both timestamps and values are compressed by default.
-            Adding this flag will keep data in an uncompressed form. Compression not only saves
-            memory but usually improve performance due to lower number of memory accesses.
+            Changes data storage from compressed (by default) to uncompressed
         labels:
             Set of label-value pairs that represent metadata labels of the key.
         chunk_size:
-            Each time-series uses chunks of memory of fixed size for time series samples.
-            You can alter the default TSDB chunk size by passing the chunk_size argument (in Bytes).
+            Memory size, in bytes, allocated for each data chunk.
 
-        For more information: https://oss.redis.com/redistimeseries/master/commands/#tsincrbytsdecrby
+        For more information: https://redis.io/commands/ts.incrby/
         """  # noqa
-        timestamp = kwargs.get("timestamp", None)
-        retention_msecs = kwargs.get("retention_msecs", None)
-        uncompressed = kwargs.get("uncompressed", False)
-        labels = kwargs.get("labels", {})
-        chunk_size = kwargs.get("chunk_size", None)
         params = [key, value]
         self._append_timestamp(params, timestamp)
         self._append_retention(params, retention_msecs)
@@ -205,12 +224,19 @@ class TimeSeriesCommands:
 
         return self.execute_command(INCRBY_CMD, *params)
 
-    def decrby(self, key, value, **kwargs):
+    def decrby(
+        self,
+        key: KeyT,
+        value: Number,
+        timestamp: Optional[Union[int, str]] = None,
+        retention_msecs: Optional[int] = None,
+        uncompressed: Optional[bool] = False,
+        labels: Optional[Dict[str, str]] = None,
+        chunk_size: Optional[int] = None,
+    ):
         """
-        Decrement (or create an time-series and decrement) the
-        latest sample's of a series.
-        This command can be used as a counter or gauge that
-        automatically gets history as a time series.
+        Decrement (or create an time-series and decrement) the latest sample's of a series.
+        This command can be used as a counter or gauge that automatically gets history as a time series.
 
         Args:
 
@@ -219,31 +245,19 @@ class TimeSeriesCommands:
         value:
             Numeric data value of the sample
         timestamp:
-            Timestamp of the sample. None can be used for automatic
-            timestamp (using the system clock).
+            Timestamp of the sample. * can be used for automatic timestamp (using the system clock).
         retention_msecs:
             Maximum age for samples compared to last event time (in milliseconds).
             If None or 0 is passed then  the series is not trimmed at all.
         uncompressed:
-            Since RedisTimeSeries v1.2, both timestamps and values are
-            compressed by default.
-            Adding this flag will keep data in an uncompressed form.
-            Compression not only saves
-            memory but usually improve performance due to lower number
-            of memory accesses.
+            Changes data storage from compressed (by default) to uncompressed
         labels:
             Set of label-value pairs that represent metadata labels of the key.
         chunk_size:
-            Each time-series uses chunks of memory of fixed size for time series samples.
-            You can alter the default TSDB chunk size by passing the chunk_size argument (in Bytes).
+            Memory size, in bytes, allocated for each data chunk.
 
-        For more information: https://oss.redis.com/redistimeseries/master/commands/#tsincrbytsdecrby
+        For more information: https://redis.io/commands/ts.decrby/
         """  # noqa
-        timestamp = kwargs.get("timestamp", None)
-        retention_msecs = kwargs.get("retention_msecs", None)
-        uncompressed = kwargs.get("uncompressed", False)
-        labels = kwargs.get("labels", {})
-        chunk_size = kwargs.get("chunk_size", None)
         params = [key, value]
         self._append_timestamp(params, timestamp)
         self._append_retention(params, retention_msecs)
@@ -253,14 +267,9 @@ class TimeSeriesCommands:
 
         return self.execute_command(DECRBY_CMD, *params)
 
-    def delete(self, key, from_time, to_time):
+    def delete(self, key: KeyT, from_time: int, to_time: int):
         """
-        Delete data points for a given timeseries and interval range
-        in the form of start and end delete timestamps.
-        The given timestamp interval is closed (inclusive), meaning start
-        and end data points will also be deleted.
-        Return the count for deleted items.
-        For more information see
+        Delete all samples between two timestamps for a given time series.
 
         Args:
 
@@ -271,68 +280,98 @@ class TimeSeriesCommands:
         to_time:
             End timestamp for the range deletion.
 
-        For more information: https://oss.redis.com/redistimeseries/master/commands/#tsdel
+        For more information: https://redis.io/commands/ts.del/
         """  # noqa
         return self.execute_command(DEL_CMD, key, from_time, to_time)
 
-    def createrule(self, source_key, dest_key, aggregation_type, bucket_size_msec):
+    def createrule(
+        self,
+        source_key: KeyT,
+        dest_key: KeyT,
+        aggregation_type: str,
+        bucket_size_msec: int,
+        align_timestamp: Optional[int] = None,
+    ):
         """
         Create a compaction rule from values added to `source_key` into `dest_key`.
-        Aggregating for `bucket_size_msec` where an `aggregation_type` can be
-        [`avg`, `sum`, `min`, `max`, `range`, `count`, `first`, `last`,
-        `std.p`, `std.s`, `var.p`, `var.s`]
 
-        For more information: https://oss.redis.com/redistimeseries/master/commands/#tscreaterule
+        Args:
+
+        source_key:
+            Key name for source time series
+        dest_key:
+            Key name for destination (compacted) time series
+        aggregation_type:
+            Aggregation type: One of the following:
+            [`avg`, `sum`, `min`, `max`, `range`, `count`, `first`, `last`, `std.p`,
+            `std.s`, `var.p`, `var.s`, `twa`]
+        bucket_size_msec:
+            Duration of each bucket, in milliseconds
+        align_timestamp:
+            Assure that there is a bucket that starts at exactly align_timestamp and
+            align all other buckets accordingly.
+
+        For more information: https://redis.io/commands/ts.createrule/
         """  # noqa
         params = [source_key, dest_key]
         self._append_aggregation(params, aggregation_type, bucket_size_msec)
+        if align_timestamp is not None:
+            params.append(align_timestamp)
 
         return self.execute_command(CREATERULE_CMD, *params)
 
-    def deleterule(self, source_key, dest_key):
+    def deleterule(self, source_key: KeyT, dest_key: KeyT):
         """
-        Delete a compaction rule.
-        For more information see
+        Delete a compaction rule from `source_key` to `dest_key`..
 
-        For more information: https://oss.redis.com/redistimeseries/master/commands/#tsdeleterule
+        For more information: https://redis.io/commands/ts.deleterule/
         """  # noqa
         return self.execute_command(DELETERULE_CMD, source_key, dest_key)
 
     def __range_params(
         self,
-        key,
-        from_time,
-        to_time,
-        count,
-        aggregation_type,
-        bucket_size_msec,
-        filter_by_ts,
-        filter_by_min_value,
-        filter_by_max_value,
-        align,
+        key: KeyT,
+        from_time: Union[int, str],
+        to_time: Union[int, str],
+        count: Optional[int],
+        aggregation_type: Optional[str],
+        bucket_size_msec: Optional[int],
+        filter_by_ts: Optional[List[int]],
+        filter_by_min_value: Optional[int],
+        filter_by_max_value: Optional[int],
+        align: Optional[Union[int, str]],
+        latest: Optional[bool],
+        bucket_timestamp: Optional[str],
+        empty: Optional[bool],
     ):
         """Create TS.RANGE and TS.REVRANGE arguments."""
         params = [key, from_time, to_time]
+        self._append_latest(params, latest)
         self._append_filer_by_ts(params, filter_by_ts)
         self._append_filer_by_value(params, filter_by_min_value, filter_by_max_value)
         self._append_count(params, count)
         self._append_align(params, align)
         self._append_aggregation(params, aggregation_type, bucket_size_msec)
+        self._append_bucket_timestamp(params, bucket_timestamp)
+        self._append_empty(params, empty)
 
         return params
 
     def range(
         self,
-        key,
-        from_time,
-        to_time,
-        count=None,
-        aggregation_type=None,
-        bucket_size_msec=0,
-        filter_by_ts=None,
-        filter_by_min_value=None,
-        filter_by_max_value=None,
-        align=None,
+        key: KeyT,
+        from_time: Union[int, str],
+        to_time: Union[int, str],
+        count: Optional[int] = None,
+        aggregation_type: Optional[str] = None,
+        bucket_size_msec: Optional[int] = 0,
+        filter_by_ts: Optional[List[int]] = None,
+        filter_by_min_value: Optional[int] = None,
+        filter_by_max_value: Optional[int] = None,
+        align: Optional[Union[int, str]] = None,
+        latest: Optional[bool] = False,
+        bucket_timestamp: Optional[str] = None,
+        empty: Optional[bool] = False,
     ):
         """
         Query a range in forward direction for a specific time-serie.
@@ -342,31 +381,34 @@ class TimeSeriesCommands:
         key:
             Key name for timeseries.
         from_time:
-            Start timestamp for the range query. - can be used to express
-            the minimum possible timestamp (0).
+            Start timestamp for the range query. - can be used to express the minimum possible timestamp (0).
         to_time:
-            End timestamp for range query, + can be used to express the
-            maximum possible timestamp.
+            End timestamp for range query, + can be used to express the maximum possible timestamp.
         count:
-            Optional maximum number of returned results.
+            Limits the number of returned samples.
         aggregation_type:
-            Optional aggregation type. Can be one of
-            [`avg`, `sum`, `min`, `max`, `range`, `count`,
-            `first`, `last`, `std.p`, `std.s`, `var.p`, `var.s`]
+            Optional aggregation type. Can be one of [`avg`, `sum`, `min`, `max`,
+            `range`, `count`, `first`, `last`, `std.p`, `std.s`, `var.p`, `var.s`, `twa`]
         bucket_size_msec:
             Time bucket for aggregation in milliseconds.
         filter_by_ts:
             List of timestamps to filter the result by specific timestamps.
         filter_by_min_value:
-            Filter result by minimum value (must mention also filter
-            by_max_value).
+            Filter result by minimum value (must mention also filter by_max_value).
         filter_by_max_value:
-            Filter result by maximum value (must mention also filter
-            by_min_value).
+            Filter result by maximum value (must mention also filter by_min_value).
         align:
             Timestamp for alignment control for aggregation.
-
-        For more information: https://oss.redis.com/redistimeseries/master/commands/#tsrangetsrevrange
+        latest:
+            Used when a time series is a compaction, reports the compacted value of the
+            latest possibly partial bucket
+        bucket_timestamp:
+            Controls how bucket timestamps are reported. Can be one of [`-`, `low`, `+`,
+            `high`, `~`, `mid`].
+        empty:
+            Reports aggregations for empty buckets.
+
+        For more information: https://redis.io/commands/ts.range/
         """  # noqa
         params = self.__range_params(
             key,
@@ -379,21 +421,27 @@ class TimeSeriesCommands:
             filter_by_min_value,
             filter_by_max_value,
             align,
+            latest,
+            bucket_timestamp,
+            empty,
         )
         return self.execute_command(RANGE_CMD, *params)
 
     def revrange(
         self,
-        key,
-        from_time,
-        to_time,
-        count=None,
-        aggregation_type=None,
-        bucket_size_msec=0,
-        filter_by_ts=None,
-        filter_by_min_value=None,
-        filter_by_max_value=None,
-        align=None,
+        key: KeyT,
+        from_time: Union[int, str],
+        to_time: Union[int, str],
+        count: Optional[int] = None,
+        aggregation_type: Optional[str] = None,
+        bucket_size_msec: Optional[int] = 0,
+        filter_by_ts: Optional[List[int]] = None,
+        filter_by_min_value: Optional[int] = None,
+        filter_by_max_value: Optional[int] = None,
+        align: Optional[Union[int, str]] = None,
+        latest: Optional[bool] = False,
+        bucket_timestamp: Optional[str] = None,
+        empty: Optional[bool] = False,
     ):
         """
         Query a range in reverse direction for a specific time-series.
@@ -409,10 +457,10 @@ class TimeSeriesCommands:
         to_time:
             End timestamp for range query, + can be used to express the maximum possible timestamp.
         count:
-            Optional maximum number of returned results.
+            Limits the number of returned samples.
         aggregation_type:
-            Optional aggregation type. Can be one of [`avg`, `sum`, `min`, `max`, `range`, `count`,
-            `first`, `last`, `std.p`, `std.s`, `var.p`, `var.s`]
+            Optional aggregation type. Can be one of [`avg`, `sum`, `min`, `max`,
+            `range`, `count`, `first`, `last`, `std.p`, `std.s`, `var.p`, `var.s`, `twa`]
         bucket_size_msec:
             Time bucket for aggregation in milliseconds.
         filter_by_ts:
@@ -423,8 +471,16 @@ class TimeSeriesCommands:
             Filter result by maximum value (must mention also filter_by_min_value).
         align:
             Timestamp for alignment control for aggregation.
-
-        For more information: https://oss.redis.com/redistimeseries/master/commands/#tsrangetsrevrange
+        latest:
+            Used when a time series is a compaction, reports the compacted value of the
+            latest possibly partial bucket
+        bucket_timestamp:
+            Controls how bucket timestamps are reported. Can be one of [`-`, `low`, `+`,
+            `high`, `~`, `mid`].
+        empty:
+            Reports aggregations for empty buckets.
+
+        For more information: https://redis.io/commands/ts.revrange/
         """  # noqa
         params = self.__range_params(
             key,
@@ -437,34 +493,43 @@ class TimeSeriesCommands:
             filter_by_min_value,
             filter_by_max_value,
             align,
+            latest,
+            bucket_timestamp,
+            empty,
         )
         return self.execute_command(REVRANGE_CMD, *params)
 
     def __mrange_params(
         self,
-        aggregation_type,
-        bucket_size_msec,
-        count,
-        filters,
-        from_time,
-        to_time,
-        with_labels,
-        filter_by_ts,
-        filter_by_min_value,
-        filter_by_max_value,
-        groupby,
-        reduce,
-        select_labels,
-        align,
+        aggregation_type: Optional[str],
+        bucket_size_msec: Optional[int],
+        count: Optional[int],
+        filters: List[str],
+        from_time: Union[int, str],
+        to_time: Union[int, str],
+        with_labels: Optional[bool],
+        filter_by_ts: Optional[List[int]],
+        filter_by_min_value: Optional[int],
+        filter_by_max_value: Optional[int],
+        groupby: Optional[str],
+        reduce: Optional[str],
+        select_labels: Optional[List[str]],
+        align: Optional[Union[int, str]],
+        latest: Optional[bool],
+        bucket_timestamp: Optional[str],
+        empty: Optional[bool],
     ):
         """Create TS.MRANGE and TS.MREVRANGE arguments."""
         params = [from_time, to_time]
+        self._append_latest(params, latest)
         self._append_filer_by_ts(params, filter_by_ts)
         self._append_filer_by_value(params, filter_by_min_value, filter_by_max_value)
+        self._append_with_labels(params, with_labels, select_labels)
         self._append_count(params, count)
         self._append_align(params, align)
         self._append_aggregation(params, aggregation_type, bucket_size_msec)
-        self._append_with_labels(params, with_labels, select_labels)
+        self._append_bucket_timestamp(params, bucket_timestamp)
+        self._append_empty(params, empty)
         params.extend(["FILTER"])
         params += filters
         self._append_groupby_reduce(params, groupby, reduce)
@@ -472,20 +537,23 @@ class TimeSeriesCommands:
 
     def mrange(
         self,
-        from_time,
-        to_time,
-        filters,
-        count=None,
-        aggregation_type=None,
-        bucket_size_msec=0,
-        with_labels=False,
-        filter_by_ts=None,
-        filter_by_min_value=None,
-        filter_by_max_value=None,
-        groupby=None,
-        reduce=None,
-        select_labels=None,
-        align=None,
+        from_time: Union[int, str],
+        to_time: Union[int, str],
+        filters: List[str],
+        count: Optional[int] = None,
+        aggregation_type: Optional[str] = None,
+        bucket_size_msec: Optional[int] = 0,
+        with_labels: Optional[bool] = False,
+        filter_by_ts: Optional[List[int]] = None,
+        filter_by_min_value: Optional[int] = None,
+        filter_by_max_value: Optional[int] = None,
+        groupby: Optional[str] = None,
+        reduce: Optional[str] = None,
+        select_labels: Optional[List[str]] = None,
+        align: Optional[Union[int, str]] = None,
+        latest: Optional[bool] = False,
+        bucket_timestamp: Optional[str] = None,
+        empty: Optional[bool] = False,
     ):
         """
         Query a range across multiple time-series by filters in forward direction.
@@ -493,46 +561,45 @@ class TimeSeriesCommands:
         Args:
 
         from_time:
-            Start timestamp for the range query. `-` can be used to
-            express the minimum possible timestamp (0).
+            Start timestamp for the range query. `-` can be used to express the minimum possible timestamp (0).
         to_time:
-            End timestamp for range query, `+` can be used to express
-            the maximum possible timestamp.
+            End timestamp for range query, `+` can be used to express the maximum possible timestamp.
         filters:
             filter to match the time-series labels.
         count:
-            Optional maximum number of returned results.
+            Limits the number of returned samples.
         aggregation_type:
-            Optional aggregation type. Can be one of
-            [`avg`, `sum`, `min`, `max`, `range`, `count`,
-            `first`, `last`, `std.p`, `std.s`, `var.p`, `var.s`]
+            Optional aggregation type. Can be one of [`avg`, `sum`, `min`, `max`,
+            `range`, `count`, `first`, `last`, `std.p`, `std.s`, `var.p`, `var.s`, `twa`]
         bucket_size_msec:
             Time bucket for aggregation in milliseconds.
         with_labels:
-            Include in the reply the label-value pairs that represent metadata
-            labels of the time-series.
-            If this argument is not set, by default, an empty Array will be
-            replied on the labels array position.
+            Include in the reply all label-value pairs representing metadata labels of the time series.
         filter_by_ts:
             List of timestamps to filter the result by specific timestamps.
         filter_by_min_value:
-            Filter result by minimum value (must mention also
-            filter_by_max_value).
+            Filter result by minimum value (must mention also filter_by_max_value).
         filter_by_max_value:
-            Filter result by maximum value (must mention also
-            filter_by_min_value).
+            Filter result by maximum value (must mention also filter_by_min_value).
         groupby:
             Grouping by fields the results (must mention also reduce).
         reduce:
-            Applying reducer functions on each group. Can be one
-            of [`sum`, `min`, `max`].
+            Applying reducer functions on each group. Can be one of [`avg` `sum`, `min`,
+            `max`, `range`, `count`, `std.p`, `std.s`, `var.p`, `var.s`].
         select_labels:
-            Include in the reply only a subset of the key-value
-            pair labels of a series.
+            Include in the reply only a subset of the key-value pair labels of a series.
         align:
             Timestamp for alignment control for aggregation.
-
-        For more information: https://oss.redis.com/redistimeseries/master/commands/#tsmrangetsmrevrange
+        latest:
+            Used when a time series is a compaction, reports the compacted
+            value of the latest possibly partial bucket
+        bucket_timestamp:
+            Controls how bucket timestamps are reported. Can be one of [`-`, `low`, `+`,
+            `high`, `~`, `mid`].
+        empty:
+            Reports aggregations for empty buckets.
+
+        For more information: https://redis.io/commands/ts.mrange/
         """  # noqa
         params = self.__mrange_params(
             aggregation_type,
@@ -549,26 +616,32 @@ class TimeSeriesCommands:
             reduce,
             select_labels,
             align,
+            latest,
+            bucket_timestamp,
+            empty,
         )
 
         return self.execute_command(MRANGE_CMD, *params)
 
     def mrevrange(
         self,
-        from_time,
-        to_time,
-        filters,
-        count=None,
-        aggregation_type=None,
-        bucket_size_msec=0,
-        with_labels=False,
-        filter_by_ts=None,
-        filter_by_min_value=None,
-        filter_by_max_value=None,
-        groupby=None,
-        reduce=None,
-        select_labels=None,
-        align=None,
+        from_time: Union[int, str],
+        to_time: Union[int, str],
+        filters: List[str],
+        count: Optional[int] = None,
+        aggregation_type: Optional[str] = None,
+        bucket_size_msec: Optional[int] = 0,
+        with_labels: Optional[bool] = False,
+        filter_by_ts: Optional[List[int]] = None,
+        filter_by_min_value: Optional[int] = None,
+        filter_by_max_value: Optional[int] = None,
+        groupby: Optional[str] = None,
+        reduce: Optional[str] = None,
+        select_labels: Optional[List[str]] = None,
+        align: Optional[Union[int, str]] = None,
+        latest: Optional[bool] = False,
+        bucket_timestamp: Optional[str] = None,
+        empty: Optional[bool] = False,
     ):
         """
         Query a range across multiple time-series by filters in reverse direction.
@@ -576,48 +649,45 @@ class TimeSeriesCommands:
         Args:
 
         from_time:
-            Start timestamp for the range query. - can be used to express
-            the minimum possible timestamp (0).
+            Start timestamp for the range query. - can be used to express the minimum possible timestamp (0).
         to_time:
-            End timestamp for range query, + can be used to express
-            the maximum possible timestamp.
+            End timestamp for range query, + can be used to express the maximum possible timestamp.
         filters:
             Filter to match the time-series labels.
         count:
-            Optional maximum number of returned results.
+            Limits the number of returned samples.
         aggregation_type:
-            Optional aggregation type. Can be one of
-            [`avg`, `sum`, `min`, `max`, `range`, `count`,
-            `first`, `last`, `std.p`, `std.s`, `var.p`, `var.s`]
+            Optional aggregation type. Can be one of [`avg`, `sum`, `min`, `max`,
+            `range`, `count`, `first`, `last`, `std.p`, `std.s`, `var.p`, `var.s`, `twa`]
         bucket_size_msec:
             Time bucket for aggregation in milliseconds.
         with_labels:
-            Include in the reply the label-value pairs that represent
-            metadata labels
-            of the time-series.
-            If this argument is not set, by default, an empty Array
-            will be replied
-            on the labels array position.
+            Include in the reply all label-value pairs representing metadata labels of the time series.
         filter_by_ts:
             List of timestamps to filter the result by specific timestamps.
         filter_by_min_value:
-            Filter result by minimum value (must mention also filter
-            by_max_value).
+            Filter result by minimum value (must mention also filter_by_max_value).
         filter_by_max_value:
-            Filter result by maximum value (must mention also filter
-            by_min_value).
+            Filter result by maximum value (must mention also filter_by_min_value).
         groupby:
             Grouping by fields the results (must mention also reduce).
         reduce:
-            Applying reducer functions on each group. Can be one
-            of [`sum`, `min`, `max`].
+            Applying reducer functions on each group. Can be one of [`avg` `sum`, `min`,
+            `max`, `range`, `count`, `std.p`, `std.s`, `var.p`, `var.s`].
         select_labels:
-            Include in the reply only a subset of the key-value pair
-            labels of a series.
+            Include in the reply only a subset of the key-value pair labels of a series.
         align:
             Timestamp for alignment control for aggregation.
-
-        For more information: https://oss.redis.com/redistimeseries/master/commands/#tsmrangetsmrevrange
+        latest:
+            Used when a time series is a compaction, reports the compacted
+            value of the latest possibly partial bucket
+        bucket_timestamp:
+            Controls how bucket timestamps are reported. Can be one of [`-`, `low`, `+`,
+            `high`, `~`, `mid`].
+        empty:
+            Reports aggregations for empty buckets.
+
+        For more information: https://redis.io/commands/ts.mrevrange/
         """  # noqa
         params = self.__mrange_params(
             aggregation_type,
@@ -634,54 +704,85 @@ class TimeSeriesCommands:
             reduce,
             select_labels,
             align,
+            latest,
+            bucket_timestamp,
+            empty,
         )
 
         return self.execute_command(MREVRANGE_CMD, *params)
 
-    def get(self, key):
+    def get(self, key: KeyT, latest: Optional[bool] = False):
         """# noqa
         Get the last sample of `key`.
+        `latest` used when a time series is a compaction, reports the compacted
+        value of the latest (possibly partial) bucket
 
-        For more information: https://oss.redis.com/redistimeseries/master/commands/#tsget
+        For more information: https://redis.io/commands/ts.get/
         """  # noqa
-        return self.execute_command(GET_CMD, key)
+        params = [key]
+        self._append_latest(params, latest)
+        return self.execute_command(GET_CMD, *params)
 
-    def mget(self, filters, with_labels=False):
+    def mget(
+        self,
+        filters: List[str],
+        with_labels: Optional[bool] = False,
+        select_labels: Optional[List[str]] = None,
+        latest: Optional[bool] = False,
+    ):
         """# noqa
         Get the last samples matching the specific `filter`.
 
-        For more information: https://oss.redis.com/redistimeseries/master/commands/#tsmget
+        Args:
+
+        filters:
+            Filter to match the time-series labels.
+        with_labels:
+            Include in the reply all label-value pairs representing metadata
+            labels of the time series.
+        select_labels:
+            Include in the reply only a subset of the key-value pair labels of a series.
+        latest:
+            Used when a time series is a compaction, reports the compacted
+            value of the latest possibly partial bucket
+
+        For more information: https://redis.io/commands/ts.mget/
         """  # noqa
         params = []
-        self._append_with_labels(params, with_labels)
+        self._append_latest(params, latest)
+        self._append_with_labels(params, with_labels, select_labels)
         params.extend(["FILTER"])
         params += filters
         return self.execute_command(MGET_CMD, *params)
 
-    def info(self, key):
+    def info(self, key: KeyT):
         """# noqa
         Get information of `key`.
 
-        For more information: https://oss.redis.com/redistimeseries/master/commands/#tsinfo
+        For more information: https://redis.io/commands/ts.info/
         """  # noqa
         return self.execute_command(INFO_CMD, key)
 
-    def queryindex(self, filters):
+    def queryindex(self, filters: List[str]):
         """# noqa
-        Get all the keys matching the `filter` list.
+        Get all time series keys matching the `filter` list.
 
-        For more information: https://oss.redis.com/redistimeseries/master/commands/#tsqueryindex
+        For more information: https://redis.io/commands/ts.queryindex/
         """  # noq
         return self.execute_command(QUERYINDEX_CMD, *filters)
 
     @staticmethod
-    def _append_uncompressed(params, uncompressed):
+    def _append_uncompressed(params: List[str], uncompressed: Optional[bool]):
         """Append UNCOMPRESSED tag to params."""
         if uncompressed:
             params.extend(["UNCOMPRESSED"])
 
     @staticmethod
-    def _append_with_labels(params, with_labels, select_labels=None):
+    def _append_with_labels(
+        params: List[str],
+        with_labels: Optional[bool],
+        select_labels: Optional[List[str]],
+    ):
         """Append labels behavior to params."""
         if with_labels and select_labels:
             raise DataError(
@@ -694,19 +795,21 @@ class TimeSeriesCommands:
             params.extend(["SELECTED_LABELS", *select_labels])
 
     @staticmethod
-    def _append_groupby_reduce(params, groupby, reduce):
+    def _append_groupby_reduce(
+        params: List[str], groupby: Optional[str], reduce: Optional[str]
+    ):
         """Append GROUPBY REDUCE property to params."""
         if groupby is not None and reduce is not None:
             params.extend(["GROUPBY", groupby, "REDUCE", reduce.upper()])
 
     @staticmethod
-    def _append_retention(params, retention):
+    def _append_retention(params: List[str], retention: Optional[int]):
         """Append RETENTION property to params."""
         if retention is not None:
             params.extend(["RETENTION", retention])
 
     @staticmethod
-    def _append_labels(params, labels):
+    def _append_labels(params: List[str], labels: Optional[List[str]]):
         """Append LABELS property to params."""
         if labels:
             params.append("LABELS")
@@ -714,38 +817,43 @@ class TimeSeriesCommands:
                 params.extend([k, v])
 
     @staticmethod
-    def _append_count(params, count):
+    def _append_count(params: List[str], count: Optional[int]):
         """Append COUNT property to params."""
         if count is not None:
             params.extend(["COUNT", count])
 
     @staticmethod
-    def _append_timestamp(params, timestamp):
+    def _append_timestamp(params: List[str], timestamp: Optional[int]):
         """Append TIMESTAMP property to params."""
         if timestamp is not None:
             params.extend(["TIMESTAMP", timestamp])
 
     @staticmethod
-    def _append_align(params, align):
+    def _append_align(params: List[str], align: Optional[Union[int, str]]):
         """Append ALIGN property to params."""
         if align is not None:
             params.extend(["ALIGN", align])
 
     @staticmethod
-    def _append_aggregation(params, aggregation_type, bucket_size_msec):
+    def _append_aggregation(
+        params: List[str],
+        aggregation_type: Optional[str],
+        bucket_size_msec: Optional[int],
+    ):
         """Append AGGREGATION property to params."""
         if aggregation_type is not None:
-            params.append("AGGREGATION")
-            params.extend([aggregation_type, bucket_size_msec])
+            params.extend(["AGGREGATION", aggregation_type, bucket_size_msec])
 
     @staticmethod
-    def _append_chunk_size(params, chunk_size):
+    def _append_chunk_size(params: List[str], chunk_size: Optional[int]):
         """Append CHUNK_SIZE property to params."""
         if chunk_size is not None:
             params.extend(["CHUNK_SIZE", chunk_size])
 
     @staticmethod
-    def _append_duplicate_policy(params, command, duplicate_policy):
+    def _append_duplicate_policy(
+        params: List[str], command: Optional[str], duplicate_policy: Optional[str]
+    ):
         """Append DUPLICATE_POLICY property to params on CREATE
         and ON_DUPLICATE on ADD.
         """
@@ -756,13 +864,33 @@ class TimeSeriesCommands:
                 params.extend(["DUPLICATE_POLICY", duplicate_policy])
 
     @staticmethod
-    def _append_filer_by_ts(params, ts_list):
+    def _append_filer_by_ts(params: List[str], ts_list: Optional[List[int]]):
         """Append FILTER_BY_TS property to params."""
         if ts_list is not None:
             params.extend(["FILTER_BY_TS", *ts_list])
 
     @staticmethod
-    def _append_filer_by_value(params, min_value, max_value):
+    def _append_filer_by_value(
+        params: List[str], min_value: Optional[int], max_value: Optional[int]
+    ):
         """Append FILTER_BY_VALUE property to params."""
         if min_value is not None and max_value is not None:
             params.extend(["FILTER_BY_VALUE", min_value, max_value])
+
+    @staticmethod
+    def _append_latest(params: List[str], latest: Optional[bool]):
+        """Append LATEST property to params."""
+        if latest:
+            params.append("LATEST")
+
+    @staticmethod
+    def _append_bucket_timestamp(params: List[str], bucket_timestamp: Optional[str]):
+        """Append BUCKET_TIMESTAMP property to params."""
+        if bucket_timestamp is not None:
+            params.extend(["BUCKETTIMESTAMP", bucket_timestamp])
+
+    @staticmethod
+    def _append_empty(params: List[str], empty: Optional[bool]):
+        """Append EMPTY property to params."""
+        if empty:
+            params.append("EMPTY")
diff --git a/redis/commands/timeseries/info.py b/redis/commands/timeseries/info.py
index fba7f093..65f3baac 100644
--- a/redis/commands/timeseries/info.py
+++ b/redis/commands/timeseries/info.py
@@ -60,15 +60,15 @@ class TSInfo:
         https://oss.redis.com/redistimeseries/configuration/#duplicate_policy
         """
         response = dict(zip(map(nativestr, args[::2]), args[1::2]))
-        self.rules = response["rules"]
-        self.source_key = response["sourceKey"]
-        self.chunk_count = response["chunkCount"]
-        self.memory_usage = response["memoryUsage"]
-        self.total_samples = response["totalSamples"]
-        self.labels = list_to_dict(response["labels"])
-        self.retention_msecs = response["retentionTime"]
-        self.lastTimeStamp = response["lastTimestamp"]
-        self.first_time_stamp = response["firstTimestamp"]
+        self.rules = response.get("rules")
+        self.source_key = response.get("sourceKey")
+        self.chunk_count = response.get("chunkCount")
+        self.memory_usage = response.get("memoryUsage")
+        self.total_samples = response.get("totalSamples")
+        self.labels = list_to_dict(response.get("labels"))
+        self.retention_msecs = response.get("retentionTime")
+        self.last_timestamp = response.get("lastTimestamp")
+        self.first_timestamp = response.get("firstTimestamp")
         if "maxSamplesPerChunk" in response:
             self.max_samples_per_chunk = response["maxSamplesPerChunk"]
             self.chunk_size = (
diff --git a/redis/ocsp.py b/redis/ocsp.py
index 4753434f..0d2c0fc4 100644
--- a/redis/ocsp.py
+++ b/redis/ocsp.py
@@ -292,7 +292,7 @@ class OCSPVerifier:
         This first retrieves for validate the certificate, issuer_url,
         and ocsp_server for certificate validate. Then retrieves the
         issuer certificate from the issuer_url, and finally checks
-        the valididy of OCSP revocation status.
+        the validity of OCSP revocation status.
         """
 
         # validate the certificate


Test patch:
diff --git a/tests/test_asyncio/test_cluster.py b/tests/test_asyncio/test_cluster.py
index f4ea5cd7..0d0ea33d 100644
--- a/tests/test_asyncio/test_cluster.py
+++ b/tests/test_asyncio/test_cluster.py
@@ -2476,3 +2476,11 @@ class TestClusterPipeline:
                         executed_on_replica = True
                         break
             assert executed_on_replica
+
+    async def test_can_run_concurrent_pipelines(self, r: RedisCluster) -> None:
+        """Test that the pipeline can be used concurrently."""
+        await asyncio.gather(
+            *(self.test_redis_cluster_pipeline(r) for i in range(100)),
+            *(self.test_multi_key_operation_with_a_single_slot(r) for i in range(100)),
+            *(self.test_multi_key_operation_with_multi_slots(r) for i in range(100)),
+        )
diff --git a/tests/test_asyncio/test_connection.py b/tests/test_asyncio/test_connection.py
index f6259adb..78a3efd2 100644
--- a/tests/test_asyncio/test_connection.py
+++ b/tests/test_asyncio/test_connection.py
@@ -1,10 +1,18 @@
 import asyncio
+import socket
 import types
+from unittest.mock import patch
 
 import pytest
 
-from redis.asyncio.connection import PythonParser, UnixDomainSocketConnection
-from redis.exceptions import InvalidResponse
+from redis.asyncio.connection import (
+    Connection,
+    PythonParser,
+    UnixDomainSocketConnection,
+)
+from redis.asyncio.retry import Retry
+from redis.backoff import NoBackoff
+from redis.exceptions import ConnectionError, InvalidResponse, TimeoutError
 from redis.utils import HIREDIS_AVAILABLE
 from tests.conftest import skip_if_server_version_lt
 
@@ -60,3 +68,44 @@ async def test_socket_param_regression(r):
 async def test_can_run_concurrent_commands(r):
     assert await r.ping() is True
     assert all(await asyncio.gather(*(r.ping() for _ in range(10))))
+
+
+async def test_connect_retry_on_timeout_error():
+    """Test that the _connect function is retried in case of a timeout"""
+    conn = Connection(retry_on_timeout=True, retry=Retry(NoBackoff(), 3))
+    origin_connect = conn._connect
+    conn._connect = mock.AsyncMock()
+
+    async def mock_connect():
+        # connect only on the last retry
+        if conn._connect.call_count <= 2:
+            raise socket.timeout
+        else:
+            return await origin_connect()
+
+    conn._connect.side_effect = mock_connect
+    await conn.connect()
+    assert conn._connect.call_count == 3
+
+
+async def test_connect_without_retry_on_os_error():
+    """Test that the _connect function is not being retried in case of a OSError"""
+    with patch.object(Connection, "_connect") as _connect:
+        _connect.side_effect = OSError("")
+        conn = Connection(retry_on_timeout=True, retry=Retry(NoBackoff(), 2))
+        with pytest.raises(ConnectionError):
+            await conn.connect()
+        assert _connect.call_count == 1
+
+
+async def test_connect_timeout_error_without_retry():
+    """Test that the _connect function is not being retried if retry_on_timeout is
+    set to False"""
+    conn = Connection(retry_on_timeout=False)
+    conn._connect = mock.AsyncMock()
+    conn._connect.side_effect = socket.timeout
+
+    with pytest.raises(TimeoutError) as e:
+        await conn.connect()
+    assert conn._connect.call_count == 1
+    assert str(e.value) == "Timeout connecting to server"
diff --git a/tests/test_asyncio/test_sentinel_managed_connection.py b/tests/test_asyncio/test_sentinel_managed_connection.py
new file mode 100644
index 00000000..a6e9f37a
--- /dev/null
+++ b/tests/test_asyncio/test_sentinel_managed_connection.py
@@ -0,0 +1,37 @@
+import socket
+
+import pytest
+
+from redis.asyncio.retry import Retry
+from redis.asyncio.sentinel import SentinelManagedConnection
+from redis.backoff import NoBackoff
+
+from .compat import mock
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_connect_retry_on_timeout_error():
+    """Test that the _connect function is retried in case of a timeout"""
+    connection_pool = mock.AsyncMock()
+    connection_pool.get_master_address = mock.AsyncMock(
+        return_value=("localhost", 6379)
+    )
+    conn = SentinelManagedConnection(
+        retry_on_timeout=True,
+        retry=Retry(NoBackoff(), 3),
+        connection_pool=connection_pool,
+    )
+    origin_connect = conn._connect
+    conn._connect = mock.AsyncMock()
+
+    async def mock_connect():
+        # connect only on the last retry
+        if conn._connect.call_count <= 2:
+            raise socket.timeout
+        else:
+            return await origin_connect()
+
+    conn._connect.side_effect = mock_connect
+    await conn.connect()
+    assert conn._connect.call_count == 3
diff --git a/tests/test_cluster.py b/tests/test_cluster.py
index 03533230..5652673a 100644
--- a/tests/test_cluster.py
+++ b/tests/test_cluster.py
@@ -672,7 +672,7 @@ class TestRedisClusterObj:
         with patch.object(Redis, "parse_response") as parse_response:
 
             def moved_redirect_effect(connection, *args, **options):
-                # raise a timeout for 5 times so we'll need to reinitilize the topology
+                # raise a timeout for 5 times so we'll need to reinitialize the topology
                 if count.val == 4:
                     parse_response.side_effect = real_func
                 count.val += 1
diff --git a/tests/test_search.py b/tests/test_search.py
index f0a1190f..642418ec 100644
--- a/tests/test_search.py
+++ b/tests/test_search.py
@@ -1698,3 +1698,17 @@ def test_dialect(modclient: redis.Redis):
     with pytest.raises(redis.ResponseError) as err:
         modclient.ft().explain(Query("@title:(@num:[0 10])").dialect(2))
     assert "Syntax error" in str(err)
+
+
+@pytest.mark.redismod
+def test_expire_while_search(modclient: redis.Redis):
+    modclient.ft().create_index((TextField("txt"),))
+    modclient.hset("hset:1", "txt", "a")
+    modclient.hset("hset:2", "txt", "b")
+    modclient.hset("hset:3", "txt", "c")
+    assert 3 == modclient.ft().search(Query("*")).total
+    modclient.pexpire("hset:2", 300)
+    for _ in range(500):
+        modclient.ft().search(Query("*")).docs[1]
+    time.sleep(1)
+    assert 2 == modclient.ft().search(Query("*")).total
diff --git a/tests/test_timeseries.py b/tests/test_timeseries.py
index 7d42147a..b4b85e17 100644
--- a/tests/test_timeseries.py
+++ b/tests/test_timeseries.py
@@ -1,8 +1,11 @@
+import math
 import time
 from time import sleep
 
 import pytest
 
+import redis
+
 from .conftest import skip_ifmodversion_lt
 
 
@@ -230,6 +233,84 @@ def test_range_advanced(client):
     assert [(0, 5.0), (5, 6.0)] == client.ts().range(
         1, 0, 10, aggregation_type="count", bucket_size_msec=10, align=5
     )
+    assert [(0, 2.5500000000000003), (10, 3.95)] == client.ts().range(
+        1, 0, 10, aggregation_type="twa", bucket_size_msec=10
+    )
+
+
+@pytest.mark.redismod
+@skip_ifmodversion_lt("1.8.0", "timeseries")
+def test_range_latest(client: redis.Redis):
+    timeseries = client.ts()
+    timeseries.create("t1")
+    timeseries.create("t2")
+    timeseries.createrule("t1", "t2", aggregation_type="sum", bucket_size_msec=10)
+    timeseries.add("t1", 1, 1)
+    timeseries.add("t1", 2, 3)
+    timeseries.add("t1", 11, 7)
+    timeseries.add("t1", 13, 1)
+    res = timeseries.range("t1", 0, 20)
+    assert res == [(1, 1.0), (2, 3.0), (11, 7.0), (13, 1.0)]
+    res = timeseries.range("t2", 0, 10)
+    assert res == [(0, 4.0)]
+    res = timeseries.range("t2", 0, 10, latest=True)
+    assert res == [(0, 4.0), (10, 8.0)]
+    res = timeseries.range("t2", 0, 9, latest=True)
+    assert res == [(0, 4.0)]
+
+
+@pytest.mark.redismod
+@skip_ifmodversion_lt("1.8.0", "timeseries")
+def test_range_bucket_timestamp(client: redis.Redis):
+    timeseries = client.ts()
+    timeseries.create("t1")
+    timeseries.add("t1", 15, 1)
+    timeseries.add("t1", 17, 4)
+    timeseries.add("t1", 51, 3)
+    timeseries.add("t1", 73, 5)
+    timeseries.add("t1", 75, 3)
+    assert [(10, 4.0), (50, 3.0), (70, 5.0)] == timeseries.range(
+        "t1", 0, 100, align=0, aggregation_type="max", bucket_size_msec=10
+    )
+    assert [(20, 4.0), (60, 3.0), (80, 5.0)] == timeseries.range(
+        "t1",
+        0,
+        100,
+        align=0,
+        aggregation_type="max",
+        bucket_size_msec=10,
+        bucket_timestamp="+",
+    )
+
+
+@pytest.mark.redismod
+@skip_ifmodversion_lt("1.8.0", "timeseries")
+def test_range_empty(client: redis.Redis):
+    timeseries = client.ts()
+    timeseries.create("t1")
+    timeseries.add("t1", 15, 1)
+    timeseries.add("t1", 17, 4)
+    timeseries.add("t1", 51, 3)
+    timeseries.add("t1", 73, 5)
+    timeseries.add("t1", 75, 3)
+    assert [(10, 4.0), (50, 3.0), (70, 5.0)] == timeseries.range(
+        "t1", 0, 100, align=0, aggregation_type="max", bucket_size_msec=10
+    )
+    res = timeseries.range(
+        "t1", 0, 100, align=0, aggregation_type="max", bucket_size_msec=10, empty=True
+    )
+    for i in range(len(res)):
+        if math.isnan(res[i][1]):
+            res[i] = (res[i][0], None)
+    assert [
+        (10, 4.0),
+        (20, None),
+        (30, None),
+        (40, None),
+        (50, 3.0),
+        (60, None),
+        (70, 5.0),
+    ] == res
 
 
 @pytest.mark.redismod
@@ -262,11 +343,87 @@ def test_rev_range(client):
     assert [(1, 10.0), (0, 1.0)] == client.ts().revrange(
         1, 0, 10, aggregation_type="count", bucket_size_msec=10, align=1
     )
+    assert [(10, 3.0), (0, 2.5500000000000003)] == client.ts().revrange(
+        1, 0, 10, aggregation_type="twa", bucket_size_msec=10
+    )
+
+
+@pytest.mark.redismod
+@skip_ifmodversion_lt("1.8.0", "timeseries")
+def test_revrange_latest(client: redis.Redis):
+    timeseries = client.ts()
+    timeseries.create("t1")
+    timeseries.create("t2")
+    timeseries.createrule("t1", "t2", aggregation_type="sum", bucket_size_msec=10)
+    timeseries.add("t1", 1, 1)
+    timeseries.add("t1", 2, 3)
+    timeseries.add("t1", 11, 7)
+    timeseries.add("t1", 13, 1)
+    res = timeseries.revrange("t2", 0, 10)
+    assert res == [(0, 4.0)]
+    res = timeseries.revrange("t2", 0, 10, latest=True)
+    assert res == [(10, 8.0), (0, 4.0)]
+    res = timeseries.revrange("t2", 0, 9, latest=True)
+    assert res == [(0, 4.0)]
+
+
+@pytest.mark.redismod
+@skip_ifmodversion_lt("1.8.0", "timeseries")
+def test_revrange_bucket_timestamp(client: redis.Redis):
+    timeseries = client.ts()
+    timeseries.create("t1")
+    timeseries.add("t1", 15, 1)
+    timeseries.add("t1", 17, 4)
+    timeseries.add("t1", 51, 3)
+    timeseries.add("t1", 73, 5)
+    timeseries.add("t1", 75, 3)
+    assert [(70, 5.0), (50, 3.0), (10, 4.0)] == timeseries.revrange(
+        "t1", 0, 100, align=0, aggregation_type="max", bucket_size_msec=10
+    )
+    assert [(20, 4.0), (60, 3.0), (80, 5.0)] == timeseries.range(
+        "t1",
+        0,
+        100,
+        align=0,
+        aggregation_type="max",
+        bucket_size_msec=10,
+        bucket_timestamp="+",
+    )
+
+
+@pytest.mark.redismod
+@skip_ifmodversion_lt("1.8.0", "timeseries")
+def test_revrange_empty(client: redis.Redis):
+    timeseries = client.ts()
+    timeseries.create("t1")
+    timeseries.add("t1", 15, 1)
+    timeseries.add("t1", 17, 4)
+    timeseries.add("t1", 51, 3)
+    timeseries.add("t1", 73, 5)
+    timeseries.add("t1", 75, 3)
+    assert [(70, 5.0), (50, 3.0), (10, 4.0)] == timeseries.revrange(
+        "t1", 0, 100, align=0, aggregation_type="max", bucket_size_msec=10
+    )
+    res = timeseries.revrange(
+        "t1", 0, 100, align=0, aggregation_type="max", bucket_size_msec=10, empty=True
+    )
+    for i in range(len(res)):
+        if math.isnan(res[i][1]):
+            res[i] = (res[i][0], None)
+    assert [
+        (70, 5.0),
+        (60, None),
+        (50, 3.0),
+        (40, None),
+        (30, None),
+        (20, None),
+        (10, 4.0),
+    ] == res
 
 
 @pytest.mark.redismod
 @pytest.mark.onlynoncluster
-def testMultiRange(client):
+def test_mrange(client):
     client.ts().create(1, labels={"Test": "This", "team": "ny"})
     client.ts().create(2, labels={"Test": "This", "Taste": "That", "team": "sf"})
     for i in range(100):
@@ -351,6 +508,31 @@ def test_multi_range_advanced(client):
     assert [(0, 5.0), (5, 6.0)] == res[0]["1"][1]
 
 
+@pytest.mark.redismod
+@pytest.mark.onlynoncluster
+@skip_ifmodversion_lt("1.8.0", "timeseries")
+def test_mrange_latest(client: redis.Redis):
+    timeseries = client.ts()
+    timeseries.create("t1")
+    timeseries.create("t2", labels={"is_compaction": "true"})
+    timeseries.create("t3")
+    timeseries.create("t4", labels={"is_compaction": "true"})
+    timeseries.createrule("t1", "t2", aggregation_type="sum", bucket_size_msec=10)
+    timeseries.createrule("t3", "t4", aggregation_type="sum", bucket_size_msec=10)
+    timeseries.add("t1", 1, 1)
+    timeseries.add("t1", 2, 3)
+    timeseries.add("t1", 11, 7)
+    timeseries.add("t1", 13, 1)
+    timeseries.add("t3", 1, 1)
+    timeseries.add("t3", 2, 3)
+    timeseries.add("t3", 11, 7)
+    timeseries.add("t3", 13, 1)
+    assert client.ts().mrange(0, 10, filters=["is_compaction=true"], latest=True) == [
+        {"t2": [{}, [(0, 4.0), (10, 8.0)]]},
+        {"t4": [{}, [(0, 4.0), (10, 8.0)]]},
+    ]
+
+
 @pytest.mark.redismod
 @pytest.mark.onlynoncluster
 @skip_ifmodversion_lt("99.99.99", "timeseries")
@@ -434,6 +616,30 @@ def test_multi_reverse_range(client):
     assert [(1, 10.0), (0, 1.0)] == res[0]["1"][1]
 
 
+@pytest.mark.redismod
+@pytest.mark.onlynoncluster
+@skip_ifmodversion_lt("1.8.0", "timeseries")
+def test_mrevrange_latest(client: redis.Redis):
+    timeseries = client.ts()
+    timeseries.create("t1")
+    timeseries.create("t2", labels={"is_compaction": "true"})
+    timeseries.create("t3")
+    timeseries.create("t4", labels={"is_compaction": "true"})
+    timeseries.createrule("t1", "t2", aggregation_type="sum", bucket_size_msec=10)
+    timeseries.createrule("t3", "t4", aggregation_type="sum", bucket_size_msec=10)
+    timeseries.add("t1", 1, 1)
+    timeseries.add("t1", 2, 3)
+    timeseries.add("t1", 11, 7)
+    timeseries.add("t1", 13, 1)
+    timeseries.add("t3", 1, 1)
+    timeseries.add("t3", 2, 3)
+    timeseries.add("t3", 11, 7)
+    timeseries.add("t3", 13, 1)
+    assert client.ts().mrevrange(
+        0, 10, filters=["is_compaction=true"], latest=True
+    ) == [{"t2": [{}, [(10, 8.0), (0, 4.0)]]}, {"t4": [{}, [(10, 8.0), (0, 4.0)]]}]
+
+
 @pytest.mark.redismod
 def test_get(client):
     name = "test"
@@ -445,6 +651,21 @@ def test_get(client):
     assert 4 == client.ts().get(name)[1]
 
 
+@pytest.mark.redismod
+@skip_ifmodversion_lt("1.8.0", "timeseries")
+def test_get_latest(client: redis.Redis):
+    timeseries = client.ts()
+    timeseries.create("t1")
+    timeseries.create("t2")
+    timeseries.createrule("t1", "t2", aggregation_type="sum", bucket_size_msec=10)
+    timeseries.add("t1", 1, 1)
+    timeseries.add("t1", 2, 3)
+    timeseries.add("t1", 11, 7)
+    timeseries.add("t1", 13, 1)
+    assert (0, 4.0) == timeseries.get("t2")
+    assert (10, 8.0) == timeseries.get("t2", latest=True)
+
+
 @pytest.mark.redismod
 @pytest.mark.onlynoncluster
 def test_mget(client):
@@ -467,6 +688,24 @@ def test_mget(client):
     assert {"Taste": "That", "Test": "This"} == res[0]["2"][0]
 
 
+@pytest.mark.redismod
+@pytest.mark.onlynoncluster
+@skip_ifmodversion_lt("1.8.0", "timeseries")
+def test_mget_latest(client: redis.Redis):
+    timeseries = client.ts()
+    timeseries.create("t1")
+    timeseries.create("t2", labels={"is_compaction": "true"})
+    timeseries.createrule("t1", "t2", aggregation_type="sum", bucket_size_msec=10)
+    timeseries.add("t1", 1, 1)
+    timeseries.add("t1", 2, 3)
+    timeseries.add("t1", 11, 7)
+    timeseries.add("t1", 13, 1)
+    assert timeseries.mget(filters=["is_compaction=true"]) == [{"t2": [{}, 0, 4.0]}]
+    assert [{"t2": [{}, 10, 8.0]}] == timeseries.mget(
+        filters=["is_compaction=true"], latest=True
+    )
+
+
 @pytest.mark.redismod
 def test_info(client):
     client.ts().create(1, retention_msecs=5, labels={"currentLabel": "currentData"})
@@ -506,7 +745,7 @@ def test_pipeline(client):
     pipeline.execute()
 
     info = client.ts().info("with_pipeline")
-    assert info.lastTimeStamp == 99
+    assert info.last_timestamp == 99
     assert info.total_samples == 100
     assert client.ts().get("with_pipeline")[1] == 99 * 1.1