Change return value type for ZPOPMAX/MIN in RESP3 (#8981)

When using RESP3, ZPOPMAX/ZPOPMIN should return nested arrays for consistency
with other commands (e.g. ZRANGE).

We do that only when COUNT argument is present (similarly to how LPOP behaves).
for reasoning see https://github.com/redis/redis/issues/8824#issuecomment-855427955

This is a breaking change only when RESP3 is used, and COUNT argument is present!

(cherry picked from commit 7f342020dc)
(cherry picked from commit caaad2d686b2af0d13fbeda414e2b70e57635b5c)

src/t_zset.c

@@ -3166,11 +3166,16 @@ void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey
     }
 
     void *arraylen_ptr = addReplyDeferredLen(c);
-    long arraylen = 0;
+    long result_count = 0;
 
     /* We emit the key only for the blocking variant. */
     if (emitkey) addReplyBulk(c,key);
 
+    /* Respond with a single (flat) array in RESP2 or if countarg is not
+     * provided (returning a single element). In RESP3, when countarg is
+     * provided, use nested array.  */
+    int use_nested_array = c->resp > 2 && countarg != NULL;
+
     /* Remove the element. */
     do {
         if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
@@ -3213,16 +3218,19 @@ void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey
         serverAssertWithInfo(c,zobj,zsetDel(zobj,ele));
         server.dirty++;
 
-        if (arraylen == 0) { /* Do this only for the first iteration. */
+        if (result_count == 0) { /* Do this only for the first iteration. */
             char *events[2] = {"zpopmin","zpopmax"};
             notifyKeyspaceEvent(NOTIFY_ZSET,events[where],key,c->db->id);
             signalModifiedKey(c,c->db,key);
         }
 
+        if (use_nested_array) {
+            addReplyArrayLen(c,2);
+        }
         addReplyBulkCBuffer(c,ele,sdslen(ele));
         addReplyDouble(c,score);
         sdsfree(ele);
-        arraylen += 2;
+        ++result_count;
 
         /* Remove the key, if indeed needed. */
         if (zsetLength(zobj) == 0) {
@@ -3232,7 +3240,10 @@ void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey
         }
     } while(--count);
 
-    setDeferredArrayLen(c,arraylen_ptr,arraylen + (emitkey != 0));
+    if (!use_nested_array) {
+        result_count *= 2;
+    }
+    setDeferredArrayLen(c,arraylen_ptr,result_count + (emitkey != 0));
 }
 
 /* ZPOPMIN key [<count>] */

tests/unit/type/zset.tcl

@@ -7,6 +7,8 @@ start_server {tags {"zset"}} {
     }
 
     proc basics {encoding} {
+        set original_max_entries [lindex [r config get zset-max-ziplist-entries] 1]
+        set original_max_value [lindex [r config get zset-max-ziplist-value] 1]
         if {$encoding == "ziplist"} {
             r config set zset-max-ziplist-entries 128
             r config set zset-max-ziplist-value 64
@@ -733,6 +735,45 @@ start_server {tags {"zset"}} {
             assert_equal 0 [r zcard z1]
             assert_equal 1 [r zcard z2]
         }
+
+        r config set zset-max-ziplist-entries $original_max_entries
+        r config set zset-max-ziplist-value $original_max_value
+
+        test "Basic ZPOP - $encoding RESP3" {
+            r hello 3
+            r del z1
+            create_zset z1 {0 a 1 b 2 c 3 d}
+            assert_equal {a 0.0} [r zpopmin z1]
+            assert_equal {d 3.0} [r zpopmax z1]
+            r hello 2
+        }
+
+        test "ZPOP with count - $encoding RESP3" {
+            r hello 3
+            r del z1
+            create_zset z1 {0 a 1 b 2 c 3 d}
+            assert_equal {{a 0.0} {b 1.0}} [r zpopmin z1 2]
+            assert_equal {{d 3.0} {c 2.0}} [r zpopmax z1 2]
+            r hello 2
+        }
+
+        test "BZPOP - $encoding RESP3" {
+            r hello 3
+            set rd [redis_deferring_client]
+            create_zset zset {0 a 1 b 2 c}
+
+            $rd bzpopmin zset 5
+            assert_equal {zset a 0} [$rd read]
+            $rd bzpopmin zset 5
+            assert_equal {zset b 1} [$rd read]
+            $rd bzpopmax zset 5
+            assert_equal {zset c 2} [$rd read]
+            assert_equal 0 [r exists zset]
+            r hello 2
+        }
+
+        r config set zset-max-ziplist-entries $original_max_entries
+        r config set zset-max-ziplist-value $original_max_value
     }
 
     basics ziplist