fix index.clear()

Before: the `.clear()` algorithm is wrong:

1. Suppose they are 2200 documents. The first 500 are deleted:

    0        1000      2000
    |        |         |
    xxxxx-----------------
    |   |
    |   index + limit
    index

2. There are now 1700 documents, and the next pagination query wrongly
   skips the first 500 documents, and remove from 500 to 1000:

    0        1000
    |        |
    -----xxxxx-------
         |   |
         |   index + limit
         index

3. There are now 1200 documents, and the next pagination query wrongly
   skips the first 1000 documents, and remove from 1000 to 1200:

    0        1000
    |        |
    ----------xx

=> Only 1200 of the 2200 documents are remove, and 1000 documents remain.

After: 500 documents are deleted in loop until there is nothing to
delete (with a security condition on the initial documents number, in
case of concurrent insertions).

Author: dfroger-idnow

redisvl/index/index.py

@@ -1,6 +1,7 @@
 import asyncio
 import json
 import threading
+from math import ceil
 import time
 import warnings
 import weakref
@@ -717,7 +718,7 @@ def _delete_batch(self, batch_keys: List[str]) -> int:
             batch_keys (List[str]): List of Redis keys to delete.
 
         Returns:
-            int: Count of records deleted from Redis.
+            int: Count of recordrecords deleted from Redis.
         """
         client = cast(SyncRedisClient, self._redis_client)
         is_cluster = isinstance(client, RedisCluster)
@@ -745,14 +746,21 @@ def clear(self) -> int:
         Returns:
             int: Count of records deleted from Redis.
         """
+        batch_size = 500
+        max_ratio = 1.01
+
+        info = self.info()
+        max_records_deleted = ceil(info["num_records"]*max_ratio)  # Allow to remove some additional concurrent inserts
         total_records_deleted: int = 0
+        query = FilterQuery(FilterExpression("*"), return_fields=["id"]).paging(0, batch_size)
 
-        for batch in self.paginate(
-            FilterQuery(FilterExpression("*"), return_fields=["id"]), page_size=500
-        ):
-            batch_keys = [record["id"] for record in batch]
-            if batch_keys:
+        while True:
+            batch = self._query(query)
+            if batch and total_records_deleted <= max_records_deleted:
+                batch_keys = [record["id"] for record in batch]
                 total_records_deleted += self._delete_batch(batch_keys)
+            else:
+                break
 
         return total_records_deleted
 
@@ -1615,14 +1623,21 @@ async def clear(self) -> int:
         Returns:
             int: Count of records deleted from Redis.
         """
+        batch_size = 500
+        max_ratio = 1.01
+
+        info = await self.info()
+        max_records_deleted = ceil(info["num_records"]*max_ratio)  # Allow to remove some additional concurrent inserts
         total_records_deleted: int = 0
+        query = FilterQuery(FilterExpression("*"), return_fields=["id"]).paging(0, batch_size)
 
-        async for batch in self.paginate(
-            FilterQuery(FilterExpression("*"), return_fields=["id"]), page_size=500
-        ):
-            batch_keys = [record["id"] for record in batch]
-            if batch_keys:
+        while True:
+            batch = await self._query(query)
+            if batch and total_records_deleted <= max_records_deleted:
+                batch_keys = [record["id"] for record in batch]
                 total_records_deleted += await self._delete_batch(batch_keys)
+            else:
+                break
 
         return total_records_deleted