chore: optimize element removal for integer lists (#3827)

Fixes #3806

Signed-off-by: Roman Gershman <roman@dragonflydb.io>

src/redis/listpack.h

@@ -76,6 +76,7 @@ unsigned char *lpGet(unsigned char *p, int64_t *count, unsigned char *intbuf);
 
 // Fills count and returns 1 if the item is an integer, 0 otherwise.
 int lpGetInteger(unsigned char *p, int64_t *ival);
+int lpStringToInt64(const char *s, unsigned long slen, int64_t *value);
 
 unsigned char *lpGetValue(unsigned char *p, unsigned int *slen, long long *lval);
 unsigned char *lpFind(unsigned char *lp, unsigned char *p, unsigned char *s, uint32_t slen, unsigned int skip);

src/redis/quicklist.c

@@ -1184,14 +1184,6 @@ int quicklistDelRange(quicklist *quicklist, const long start, const long count)
     return 1;
 }
 
-/* compare between a two entries */
-int quicklistCompare(const quicklistEntry* entry, const unsigned char *p2, const size_t p2_len) {
-    if (unlikely(QL_NODE_IS_PLAIN(entry->node))) {
-        return ((entry->sz == p2_len) && (memcmp(entry->value, p2, p2_len) == 0));
-    }
-    return lpCompare(entry->zi, p2, p2_len);
-}
-
 /* Returns a quicklist iterator 'iter'. After the initialization every
  * call to quicklistNext() will return the next element of the quicklist. */
 quicklistIter *quicklistGetIterator(quicklist *quicklist, int direction) {

src/redis/quicklist.h

@@ -186,7 +186,6 @@ int quicklistPopCustom(quicklist *quicklist,
                        void *(*saver)(unsigned char *data, size_t sz));
 int quicklistPop(quicklist *quicklist, int where, unsigned char **data, size_t *sz, long long *slong);
 unsigned long quicklistCount(const quicklist *ql);
-int quicklistCompare(const quicklistEntry *entry, const unsigned char *p2, const size_t p2_len);
 size_t quicklistGetLzf(const quicklistNode *node, void **data);
 void quicklistNodeLimit(int fill, size_t *size, unsigned int *count);
 int quicklistNodeExceedsLimit(int fill, size_t new_sz, unsigned int new_count);

src/server/list_family.cc

@@ -547,10 +547,20 @@ OpResult<uint32_t> OpRem(const OpArgs& op_args, string_view key, string_view ele
   quicklistInitIterator(&qiter, ql, iter_direction, index);
   quicklistEntry entry;
   unsigned removed = 0;
-  const uint8_t* elem_ptr = reinterpret_cast<const uint8_t*>(elem.data());
+  int64_t ival;
+
+  // try parsing the element into an integer.
+  int is_int = lpStringToInt64(elem.data(), elem.size(), &ival);
+
+  auto is_match = [&](const quicklistEntry& entry) {
+    if (is_int != (entry.value == nullptr))
+      return false;
+
+    return is_int ? entry.longval == ival : ElemCompare(entry, elem);
+  };
 
   while (quicklistNext(&qiter, &entry)) {
-    if (quicklistCompare(&entry, elem_ptr, elem.size())) {
+    if (is_match(entry)) {
       quicklistDelEntry(&qiter, &entry);
       removed++;
       if (count && removed == count)

src/server/list_family_test.cc

@@ -370,6 +370,13 @@ TEST_F(ListFamilyTest, LRem) {
   Run({"set", "foo", "bar"});
   ASSERT_THAT(Run({"lrem", "foo", "0", "elem"}), ErrArg("WRONGTYPE"));
   ASSERT_THAT(Run({"lrem", "nexists", "0", "elem"}), IntArg(0));
+
+  // Triggers QUICKLIST_NODE_CONTAINER_PLAIN coverage
+  string val(10000, 'a');
+  Run({"rpush", kKey2, val, "12345678"});
+
+  ASSERT_THAT(Run({"lrem", kKey2, "1", "12345678"}), IntArg(1));
+  ASSERT_THAT(Run({"lrem", kKey2, "1", val}), IntArg(1));
 }
 
 TEST_F(ListFamilyTest, LTrim) {