diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a8eea21c..e4449c1d6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,28 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **Six sorted-set commands that were `unknown command`, and `ZADD ... INCR`** + (moon#959). `ZRANGEBYLEX`, `ZREVRANGEBYLEX`, `ZREMRANGEBYRANK`, + `ZREMRANGEBYSCORE`, `ZREMRANGEBYLEX` and `ZDIFFSTORE` are implemented, wired + into every dispatch path, registered as `@sortedset`, and covered by rows in + both parity harnesses; `docs/commands.md` had advertised `ZRANGEBYLEX` while + dispatch rejected it. `ZADD ... INCR` — which `redis-py`'s `zadd(..., + incr=True)` sends — replies the new score as a bulk string, or nil when + `NX`/`XX`/`GT`/`LT` refuse, in Redis's decision order. Every reply, error + surface included, was read off redis-server 8.6.1 before the code was + written: the range grammar is checked before the key is consulted, a + `ZREMRANGEBY*` that drains a key deletes it, a listpack zset is trimmed in + place and never converted, and the `used_memory` ledger stays exact on both + encodings. `ZDIFFSTORE` joins the `ZUNIONSTORE` family's `numkeys` and + option rules, refusing `WEIGHTS`/`AGGREGATE` as `syntax error`, and — because + it writes a destination it is not routed on — it also joins the moon#592 + cross-shard WRITE guard, so `ZDIFFSTORE` across shards is `CROSSSLOT` rather + than an ack whose destination lands nowhere. (It shares the guard's + `(10, b'z')` match arm with moon#962's `ZINTERCARD`; both spellings are + named there.) + ### Changed - **`Check (macOS)` and `Check (Windows)` run their tests in three shards**, @@ -276,6 +298,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 its two root pages at the real path, which every later open rejected. Tombstones are also aged per `(file_id, file_type)`, not per id. +- **`ZUNIONSTORE`/`ZINTERSTORE` report `WRONGTYPE` before an option error, and + no longer flatten a listpack source** (moon#959). Redis looks every source up + before it parses `WEIGHTS`/`AGGREGATE`, so `ZUNIONSTORE d 1 + BOGUS` is `WRONGTYPE` on redis 8.6.1; moon answered `syntax error`. The store + family also read its sources through the promoting accessor, converting a + `listpack` source to `skiplist` as a side effect of reading it — the moon#928 + defect the read-only set operations were already cured of. Both fixes came + with the shared implementation `ZDIFFSTORE` now uses. - **Commands routed to another shard are counted and timed** (moon#982). At `--shards > 1` a command whose key lives on a shard other than the connection's went through no telemetry probe at all — neither the diff --git a/docs/commands.md b/docs/commands.md index 35dded015..888e94e62 100644 --- a/docs/commands.md +++ b/docs/commands.md @@ -47,9 +47,12 @@ Per-field return code on the TTL commands: `-2` = no such field, `-1` = no TTL, `SADD`, `SREM`, `SMEMBERS`, `SCARD`, `SISMEMBER`, `SMISMEMBER`, `SINTER`, `SUNION`, `SDIFF`, `SINTERSTORE`, `SUNIONSTORE`, `SDIFFSTORE`, `SRANDMEMBER`, `SPOP`, `SSCAN` -## Sorted sets (21) +## Sorted sets (35) -`ZADD`, `ZREM`, `ZSCORE`, `ZCARD`, `ZINCRBY`, `ZRANK`, `ZREVRANK`, `ZPOPMIN`, `ZPOPMAX`, `ZSCAN`, `ZRANGE`, `ZREVRANGE`, `ZRANGEBYSCORE`, `ZREVRANGEBYSCORE`, `ZRANGEBYLEX`, `ZCOUNT`, `ZLEXCOUNT`, `ZUNIONSTORE`, `ZINTERSTORE`, `BZPOPMIN`, `BZPOPMAX` +`ZADD`, `ZREM`, `ZSCORE`, `ZCARD`, `ZINCRBY`, `ZRANK`, `ZREVRANK`, `ZPOPMIN`, `ZPOPMAX`, `ZSCAN`, `ZRANGE`, `ZREVRANGE`, `ZRANGEBYSCORE`, `ZREVRANGEBYSCORE`, `ZRANGEBYLEX`, `ZREVRANGEBYLEX`, `ZCOUNT`, `ZLEXCOUNT`, `ZREMRANGEBYRANK`, `ZREMRANGEBYSCORE`, `ZREMRANGEBYLEX`, `ZUNIONSTORE`, `ZINTERSTORE`, `ZDIFFSTORE`, `ZRANGESTORE`, `ZDIFF`, `ZUNION`, `ZINTER`, `ZINTERCARD`, `ZMSCORE`, `ZRANDMEMBER`, `ZMPOP`, `BZPOPMIN`, `BZPOPMAX`, `BZMPOP` + +!!! tip + `ZADD` supports `NX`, `XX`, `GT`, `LT`, `CH` and `INCR`, matching Redis 6.2+ behavior. Every command in this list was sent to a live moon server and answered (moon#959 re-verified the table command by command, not from `COMMAND INFO`, after it had advertised an unimplemented one). ## Geospatial (8) diff --git a/scripts/test-commands.sh b/scripts/test-commands.sh index b5c1c62a7..c498c9229 100755 --- a/scripts/test-commands.sh +++ b/scripts/test-commands.sh @@ -952,6 +952,45 @@ if should_run "sorted_set"; then rcli ZADD z:ch 1 m >/dev/null 2>&1; mcli ZADD z:ch 1 m >/dev/null 2>&1 assert_match "ZADD CH sub-epsilon" ZADD z:ch CH 1.0000000000000002 m assert_match "ZADD CH moved the score" ZSCORE z:ch m + + # moon#959 -- six commands that answered `ERR unknown command` on moon + # (ZRANGEBYLEX, ZREVRANGEBYLEX, ZREMRANGEBYRANK, ZREMRANGEBYSCORE, + # ZREMRANGEBYLEX, ZDIFFSTORE) plus `ZADD ... INCR`, which answered an + # arity error. Neither harness named any of them, which is how a command + # docs/commands.md advertised went missing. Every reply was read off + # redis 8.6.1 before the commands were written. + rcli ZADD z:959:lex 0 a 0 b 0 c 0 d 0 e >/dev/null 2>&1; mcli ZADD z:959:lex 0 a 0 b 0 c 0 d 0 e >/dev/null 2>&1 + assert_match "ZRANGEBYLEX" ZRANGEBYLEX z:959:lex - + + assert_match "ZRANGEBYLEX bounds" ZRANGEBYLEX z:959:lex '[b' '(d' + assert_match "ZRANGEBYLEX LIMIT" ZRANGEBYLEX z:959:lex - + LIMIT 1 2 + assert_match "ZRANGEBYLEX bad bound" ZRANGEBYLEX z:959:lex a b + assert_match "ZRANGEBYLEX WITHSCORES" ZRANGEBYLEX z:959:lex - + WITHSCORES + assert_match "ZRANGEBYLEX WITHSCORES 1st" ZRANGEBYLEX z:959:lex a b WITHSCORES + assert_match "ZREVRANGEBYLEX" ZREVRANGEBYLEX z:959:lex + - + assert_match "ZREVRANGEBYLEX bounds" ZREVRANGEBYLEX z:959:lex '(d' '[b' LIMIT 0 1 + rcli ZADD z:959:r 1 a 2 b 3 c 4 d 5 e >/dev/null 2>&1; mcli ZADD z:959:r 1 a 2 b 3 c 4 d 5 e >/dev/null 2>&1 + assert_match "ZREMRANGEBYRANK" ZREMRANGEBYRANK z:959:r 0 0 + assert_match "ZREMRANGEBYRANK neg stop" ZREMRANGEBYRANK z:959:r -10 -6 + assert_match "ZREMRANGEBYSCORE" ZREMRANGEBYSCORE z:959:r '(2' 3 + assert_match "ZREMRANGEBYSCORE bad" ZREMRANGEBYSCORE z:959:r nan 1 + assert_match "ZREMRANGEBYLEX" ZREMRANGEBYLEX z:959:lex '[b' '(d' + assert_match "ZREMRANGEBYLEX arity" ZREMRANGEBYLEX z:959:lex - + x + assert_match "ZRANGE after ZREMRANGE" ZRANGE z:959:r 0 -1 WITHSCORES + assert_match "ZREMRANGEBYSCORE drains" ZREMRANGEBYSCORE z:959:r -inf +inf + assert_match "ZREMRANGE drained key" EXISTS z:959:r + assert_match "ZDIFFSTORE" ZDIFFSTORE {z}:diff 2 {z}:A {z}:B + assert_match "ZDIFFSTORE result" ZRANGE {z}:diff 0 -1 WITHSCORES + assert_match "ZDIFFSTORE numkeys 0" ZDIFFSTORE {z}:diff 0 {z}:A + assert_match "ZDIFFSTORE WEIGHTS" ZDIFFSTORE {z}:diff 1 {z}:A WEIGHTS 1 + assert_match "ZDIFFSTORE empty deletes" ZDIFFSTORE {z}:diff 2 {z}:A {z}:A + assert_match "ZDIFFSTORE dest gone" EXISTS {z}:diff + assert_match "ZADD INCR" ZADD z:959:i INCR 5 a + assert_match "ZADD INCR again" ZADD z:959:i INCR 2.5 a + assert_match "ZADD NX INCR refused" ZADD z:959:i NX INCR 1 a + assert_match "ZADD XX INCR refused" ZADD z:959:i XX INCR 1 nope + assert_match "ZADD GT INCR refused" ZADD z:959:i GT INCR -1 a + assert_match "ZADD LT INCR" ZADD z:959:i LT INCR -1 a + assert_match "ZADD INCR two pairs" ZADD z:959:i INCR 1 a 2 b fi # =========================================================================== diff --git a/scripts/test-consistency.sh b/scripts/test-consistency.sh index f28cd341f..81a0c6817 100755 --- a/scripts/test-consistency.sh +++ b/scripts/test-consistency.sh @@ -952,6 +952,122 @@ both ZADD z:792:bt 1 bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb assert_both "ZADD CH sub-epsilon (bptree)" ZADD z:792:bt CH 1.0000000000000002 bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb assert_both "ZADD CH bptree moved score" ZSCORE z:792:bt bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb +# moon#959 -- six commands that answered `ERR unknown command` on moon +# (ZRANGEBYLEX, ZREVRANGEBYLEX, ZREMRANGEBYRANK, ZREMRANGEBYSCORE, +# ZREMRANGEBYLEX, ZDIFFSTORE) plus `ZADD ... INCR`, which answered an arity +# error. Every reply was read off redis 8.6.1 before the commands were +# written, error surface included: the bounds grammar is checked BEFORE the +# key (a bad bound on a missing key is an error, not an empty array), a +# drained key is deleted, and WRONGTYPE never clobbers the value it refused. +both ZADD z:959:lex 0 a 0 b 0 c 0 d 0 e +assert_both "ZRANGEBYLEX all" ZRANGEBYLEX z:959:lex - + +assert_both "ZRANGEBYLEX [b (d" ZRANGEBYLEX z:959:lex '[b' '(d' +assert_both "ZRANGEBYLEX LIMIT 1 2" ZRANGEBYLEX z:959:lex - + LIMIT 1 2 +assert_both "ZRANGEBYLEX LIMIT -1 2" ZRANGEBYLEX z:959:lex - + LIMIT -1 2 +assert_both "ZRANGEBYLEX reversed bounds" ZRANGEBYLEX z:959:lex + - +assert_both "ZRANGEBYLEX bad bound" ZRANGEBYLEX z:959:lex a b +assert_both "ZRANGEBYLEX bad bound missing key" ZRANGEBYLEX z:959:nokey a b +assert_both "ZRANGEBYLEX WITHSCORES" ZRANGEBYLEX z:959:lex - + WITHSCORES +assert_both "ZRANGEBYLEX WITHSCORES beats bound" ZRANGEBYLEX z:959:lex a b WITHSCORES +assert_both "ZRANGEBYLEX LIMIT beats WITHSCORES" ZRANGEBYLEX z:959:lex - + WITHSCORES LIMIT 1 +assert_both "ZRANGEBYLEX dangling LIMIT" ZRANGEBYLEX z:959:lex - + LIMIT 1 +assert_both "ZRANGEBYLEX LIMIT notanint" ZRANGEBYLEX z:959:lex - + LIMIT notanint 1 +assert_both "ZRANGEBYLEX unknown token" ZRANGEBYLEX z:959:lex - + BOGUS +assert_both "ZRANGEBYLEX missing key" ZRANGEBYLEX z:959:nokey - + +assert_both "ZREVRANGEBYLEX all" ZREVRANGEBYLEX z:959:lex + - +assert_both "ZREVRANGEBYLEX (d [b" ZREVRANGEBYLEX z:959:lex '(d' '[b' +assert_both "ZREVRANGEBYLEX LIMIT" ZREVRANGEBYLEX z:959:lex + - LIMIT 1 2 +assert_both "ZREVRANGEBYLEX reversed bounds" ZREVRANGEBYLEX z:959:lex - + +both ZADD z:959:rank 1 a 2 b 3 c 4 d 5 e +assert_both "ZREMRANGEBYRANK 0 0" ZREMRANGEBYRANK z:959:rank 0 0 +# A stop still negative after normalisation is NOT clamped to 0 -- nothing +# is removed. (ZRANGE's own helper clamps it; that divergence is out of +# moon#959's scope and is reported separately.) +assert_both "ZREMRANGEBYRANK -10 -6" ZREMRANGEBYRANK z:959:rank -10 -6 +assert_both "ZREMRANGEBYRANK 3 1" ZREMRANGEBYRANK z:959:rank 3 1 +assert_both "ZREMRANGEBYRANK 1 -2" ZREMRANGEBYRANK z:959:rank 1 -2 +assert_both "ZREMRANGEBYRANK left" ZRANGE z:959:rank 0 -1 WITHSCORES +assert_both "ZREMRANGEBYRANK notanint" ZREMRANGEBYRANK z:959:rank notanint 1 +assert_both "ZREMRANGEBYRANK arity" ZREMRANGEBYRANK z:959:rank 1 +assert_both "ZREMRANGEBYRANK missing key" ZREMRANGEBYRANK z:959:nokey 0 1 +assert_both "ZREMRANGEBYRANK drains" ZREMRANGEBYRANK z:959:rank 0 -1 +assert_both "ZREMRANGEBYRANK drained key gone" EXISTS z:959:rank +both ZADD z:959:score 1 a 2 b 3 c 4 d 5 e +assert_both "ZREMRANGEBYSCORE (2 3" ZREMRANGEBYSCORE z:959:score '(2' 3 +assert_both "ZREMRANGEBYSCORE 3 1" ZREMRANGEBYSCORE z:959:score 3 1 +assert_both "ZREMRANGEBYSCORE 5 inf" ZREMRANGEBYSCORE z:959:score 5 inf +assert_both "ZREMRANGEBYSCORE left" ZRANGE z:959:score 0 -1 WITHSCORES +assert_both "ZREMRANGEBYSCORE nan" ZREMRANGEBYSCORE z:959:score nan 1 +assert_both "ZREMRANGEBYSCORE bad on missing" ZREMRANGEBYSCORE z:959:nokey a 1 +assert_both "ZREMRANGEBYSCORE drains" ZREMRANGEBYSCORE z:959:score -inf +inf +assert_both "ZREMRANGEBYSCORE drained key gone" EXISTS z:959:score +both ZADD z:959:lex2 0 a 0 b 0 c 0 d 0 e +assert_both "ZREMRANGEBYLEX [b (d" ZREMRANGEBYLEX z:959:lex2 '[b' '(d' +assert_both "ZREMRANGEBYLEX (c +" ZREMRANGEBYLEX z:959:lex2 '(c' + +assert_both "ZREMRANGEBYLEX left" ZRANGE z:959:lex2 0 -1 +assert_both "ZREMRANGEBYLEX bad bound" ZREMRANGEBYLEX z:959:lex2 a b +assert_both "ZREMRANGEBYLEX arity" ZREMRANGEBYLEX z:959:lex2 - + x +assert_both "ZREMRANGEBYLEX drains" ZREMRANGEBYLEX z:959:lex2 - + +assert_both "ZREMRANGEBYLEX drained key gone" EXISTS z:959:lex2 +both SET z:959:str v +assert_both "ZRANGEBYLEX WRONGTYPE" ZRANGEBYLEX z:959:str - + +assert_both "ZREVRANGEBYLEX WRONGTYPE" ZREVRANGEBYLEX z:959:str + - +assert_both "ZREMRANGEBYRANK WRONGTYPE" ZREMRANGEBYRANK z:959:str 0 1 +assert_both "ZREMRANGEBYSCORE WRONGTYPE" ZREMRANGEBYSCORE z:959:str 0 1 +assert_both "ZREMRANGEBYLEX WRONGTYPE" ZREMRANGEBYLEX z:959:str - + +assert_both "WRONGTYPE left the string" GET z:959:str +# ZDIFFSTORE joins the ZUNIONSTORE family: the same two numkeys classes, the +# same overrun rule, and EVERY option token refused (it takes none). Redis +# looks the sources up before it parses the options, so WRONGTYPE outranks +# an option error on all three STORE commands. `{z959}` co-locates the +# destination with its sources (moon#592). +both ZADD {z959}:a 1 a 2 b 3 c 4 d 5 e +both ZADD {z959}:b 1 a 2 b +both ZADD {z959}:c 2 b 9 x +both SET {z959}:str v +assert_both "ZDIFFSTORE two sources" ZDIFFSTORE {z959}:diff 2 {z959}:a {z959}:b +assert_both "ZDIFFSTORE result" ZRANGE {z959}:diff 0 -1 WITHSCORES +assert_both "ZDIFFSTORE three sources" ZDIFFSTORE {z959}:diff 3 {z959}:a {z959}:b {z959}:c +assert_both "ZDIFFSTORE result 3" ZRANGE {z959}:diff 0 -1 WITHSCORES +assert_both "ZDIFFSTORE missing first source" ZDIFFSTORE {z959}:diff 2 {z959}:nokey {z959}:a +assert_both "ZDIFFSTORE empty deletes dest" EXISTS {z959}:diff +assert_both "ZDIFFSTORE dest is a source" ZDIFFSTORE {z959}:c 2 {z959}:a {z959}:c +assert_both "ZDIFFSTORE dest-as-source result" ZRANGE {z959}:c 0 -1 WITHSCORES +assert_both "ZDIFFSTORE numkeys 0" ZDIFFSTORE {z959}:diff 0 {z959}:a +assert_both "ZDIFFSTORE numkeys -1" ZDIFFSTORE {z959}:diff -1 {z959}:a +assert_both "ZDIFFSTORE numkeys notanint" ZDIFFSTORE {z959}:diff notanint {z959}:a +assert_both "ZDIFFSTORE numkeys overruns" ZDIFFSTORE {z959}:diff 2 {z959}:a +assert_both "ZDIFFSTORE WEIGHTS refused" ZDIFFSTORE {z959}:diff 1 {z959}:a WEIGHTS 1 +assert_both "ZDIFFSTORE AGGREGATE refused" ZDIFFSTORE {z959}:diff 1 {z959}:a AGGREGATE SUM +assert_both "ZDIFFSTORE unknown token" ZDIFFSTORE {z959}:diff 1 {z959}:a BOGUS +assert_both "ZDIFFSTORE arity" ZDIFFSTORE {z959}:diff 1 +assert_both "ZDIFFSTORE WRONGTYPE source" ZDIFFSTORE {z959}:diff 2 {z959}:a {z959}:str +assert_both "ZDIFFSTORE WRONGTYPE beats option" ZDIFFSTORE {z959}:diff 1 {z959}:str BOGUS +assert_both "ZUNIONSTORE WRONGTYPE beats option" ZUNIONSTORE {z959}:diff 1 {z959}:str BOGUS +assert_both "ZINTERSTORE WRONGTYPE beats WEIGHTS" ZINTERSTORE {z959}:diff 2 {z959}:a {z959}:str WEIGHTS 1 1 +assert_both "ZDIFFSTORE errors made no dest" EXISTS {z959}:diff +# ZADD ... INCR: ZINCRBY's arithmetic under ZADD's flags, the new score as +# a bulk string, nil when a flag refuses. +assert_both "ZADD INCR new member" ZADD z:959:incr INCR 5 a +assert_both "ZADD INCR existing" ZADD z:959:incr INCR 2.5 a +assert_both "ZADD NX INCR present" ZADD z:959:incr NX INCR 1 a +assert_both "ZADD NX INCR absent" ZADD z:959:incr NX INCR 1 n +assert_both "ZADD XX INCR absent" ZADD z:959:incr XX INCR 1 nope +assert_both "ZADD XX INCR present" ZADD z:959:incr XX INCR 1 a +assert_both "ZADD GT INCR refused" ZADD z:959:incr GT INCR -1 a +assert_both "ZADD GT INCR zero refused" ZADD z:959:incr GT INCR 0 a +assert_both "ZADD LT INCR" ZADD z:959:incr LT INCR -1 a +assert_both "ZADD XX GT INCR absent" ZADD z:959:incr XX GT INCR 1 q +assert_both "ZADD INCR CH" ZADD z:959:incr INCR CH 1 a +assert_both "ZADD INCR two pairs" ZADD z:959:incr INCR 1 a 2 b +assert_both "ZADD INCR odd tail" ZADD z:959:incr INCR 1 +assert_both "ZADD INCR nan" ZADD z:959:incr INCR nan a +assert_both "ZADD INCR inf" ZADD z:959:incr INCR inf a +assert_both "ZADD INCR inf + -inf" ZADD z:959:incr INCR -inf a +assert_both "ZADD INCR after refusals" ZRANGE z:959:incr 0 -1 WITHSCORES +assert_both "ZADD XX INCR on missing key" ZADD z:959:incr:xx XX INCR 1 a +assert_both "ZADD XX INCR made no key" EXISTS z:959:incr:xx + # Exactly zset-max-listpack-entries (128) members is STILL a listpack; one # more promotes to a skiplist on both. One ZADD per step, not 129 — each # `both` spawns two redis-cli processes. @@ -1826,6 +1942,11 @@ if [[ "$SHARDS" -gt 1 ]]; then "zrangestore|ZADD %S 1 a 2 b|ZRANGESTORE %D %S 0 -1|ZCARD %D" "zunionstore|ZADD %S 1 a 2 b|ZUNIONSTORE %D 1 %S|ZCARD %D" "zinterstore|ZADD %S 1 a 2 b|ZINTERSTORE %D 1 %S|ZCARD %D" + # moon#959: ZDIFFSTORE joined the family the moment it stopped being + # `unknown command` -- same shape as its two siblings above, routed on + # the destination and reading every source. Without the guard arm it + # acks :2 and the destination is empty on a normally-routed read. + "zdiffstore|ZADD %S 1 a 2 b|ZDIFFSTORE %D 1 %S|ZCARD %D" "pfmerge|PFADD %S a b c|PFMERGE %D %S|PFCOUNT %D" "geosearchstore|GEOADD %S 15 37 Here|GEOSEARCHSTORE %D %S FROMLONLAT 15 37 BYRADIUS 200 km ASC|ZCARD %D" "sortstore|RPUSH %S 3 1 2|SORT %S STORE %D|LLEN %D" diff --git a/src/command/metadata.rs b/src/command/metadata.rs index a60b37db7..7089776bb 100644 --- a/src/command/metadata.rs +++ b/src/command/metadata.rs @@ -327,6 +327,14 @@ pub static COMMAND_META: phf::Map<&'static str, CommandMeta> = phf_map! { "ZMSCORE" => CommandMeta { name: "ZMSCORE", arity: -3, flags: RFP, first_key: 1, last_key: 1, step: 1, acl_categories: ZST }, "ZRANDMEMBER" => CommandMeta { name: "ZRANDMEMBER", arity: -2, flags: RP, first_key: 1, last_key: 1, step: 1, acl_categories: ZST }, "ZMPOP" => CommandMeta { name: "ZMPOP", arity: -4, flags: WP, first_key: 0, last_key: 0, step: 0, acl_categories: ZST }, + // moon#959. Arities match redis 8.6.1 as SENT — every arity error below + // was compared on the wire, not read off `COMMAND INFO`. + "ZRANGEBYLEX" => CommandMeta { name: "ZRANGEBYLEX", arity: -4, flags: RP, first_key: 1, last_key: 1, step: 1, acl_categories: ZST }, + "ZREVRANGEBYLEX" => CommandMeta { name: "ZREVRANGEBYLEX", arity: -4, flags: RP, first_key: 1, last_key: 1, step: 1, acl_categories: ZST }, + "ZREMRANGEBYRANK" => CommandMeta { name: "ZREMRANGEBYRANK", arity: 4, flags: WP, first_key: 1, last_key: 1, step: 1, acl_categories: ZST }, + "ZREMRANGEBYSCORE" => CommandMeta { name: "ZREMRANGEBYSCORE", arity: 4, flags: WP, first_key: 1, last_key: 1, step: 1, acl_categories: ZST }, + "ZREMRANGEBYLEX" => CommandMeta { name: "ZREMRANGEBYLEX", arity: 4, flags: WP, first_key: 1, last_key: 1, step: 1, acl_categories: ZST }, + "ZDIFFSTORE" => CommandMeta { name: "ZDIFFSTORE", arity: -4, flags: WP, first_key: 1, last_key: 1, step: 1, acl_categories: ZST }, // ---- Stream commands ---- "XADD" => CommandMeta { name: "XADD", arity: -5, flags: WFP, first_key: 1, last_key: 1, step: 1, acl_categories: STM }, @@ -1560,6 +1568,10 @@ mod tests { b"ZINTERSTORE", b"ZRANGESTORE", b"ZMPOP", + b"ZREMRANGEBYRANK", + b"ZREMRANGEBYSCORE", + b"ZREMRANGEBYLEX", + b"ZDIFFSTORE", b"HINCRBYFLOAT", b"LSET", b"LREM", @@ -1597,6 +1609,8 @@ mod tests { b"SMEMBERS", b"SISMEMBER", b"ZRANGEBYSCORE", + b"ZRANGEBYLEX", + b"ZREVRANGEBYLEX", b"BITFIELD_RO", b"SORT_RO", b"GEORADIUS_RO", diff --git a/src/command/mod.rs b/src/command/mod.rs index 54c3b55bc..a11cae0a4 100644 --- a/src/command/mod.rs +++ b/src/command/mod.rs @@ -996,6 +996,9 @@ fn dispatch_inner( if cmd.eq_ignore_ascii_case(b"ZINTERCARD") { return resp(sorted_set::zintercard(db, args)); } + if cmd.eq_ignore_ascii_case(b"ZDIFFSTORE") { + return resp(sorted_set::zdiffstore(db, args)); + } } // 11-letter commands (11, b'p') => { @@ -1042,6 +1045,9 @@ fn dispatch_inner( if cmd.eq_ignore_ascii_case(b"ZRANDMEMBER") { return resp(sorted_set::zrandmember(db, args)); } + if cmd.eq_ignore_ascii_case(b"ZRANGEBYLEX") { + return resp(sorted_set::zrangebylex(db, args)); + } } // 11-letter commands (hash) (11, b'h') => { @@ -1074,6 +1080,15 @@ fn dispatch_inner( } } // 14-letter commands + (14, b'z') => { + // ZREVRANGEBYLEX ZREMRANGEBYLEX + if cmd.eq_ignore_ascii_case(b"ZREVRANGEBYLEX") { + return resp(sorted_set::zrevrangebylex(db, args)); + } + if cmd.eq_ignore_ascii_case(b"ZREMRANGEBYLEX") { + return resp(sorted_set::zremrangebylex(db, args)); + } + } (14, b'g') => { // GEOSEARCHSTORE if cmd.eq_ignore_ascii_case(b"GEOSEARCHSTORE") { @@ -1086,6 +1101,16 @@ fn dispatch_inner( if cmd.eq_ignore_ascii_case(b"ZREVRANGEBYSCORE") { return resp(sorted_set::zrevrangebyscore(db, args)); } + if cmd.eq_ignore_ascii_case(b"ZREMRANGEBYSCORE") { + return resp(sorted_set::zremrangebyscore(db, args)); + } + } + // 15-letter commands + (15, b'z') => { + // ZREMRANGEBYRANK + if cmd.eq_ignore_ascii_case(b"ZREMRANGEBYRANK") { + return resp(sorted_set::zremrangebyrank(db, args)); + } } // 17-letter commands (17, b'g') => { @@ -1208,6 +1233,7 @@ pub fn is_dispatch_read_supported(cmd: &[u8]) -> bool { | (12, b'h') // HPEXPIRETIME | (12, b'g') // GEORADIUS_RO | (13, b'z') // ZRANGEBYSCORE + | (14, b'z') // ZREVRANGEBYLEX | (16, b'z') // ZREVRANGEBYSCORE | (20, b'g') // GEORADIUSBYMEMBER_RO ) @@ -1615,6 +1641,12 @@ fn dispatch_read_inner(db: &Database, cmd: &[u8], args: &[Frame], now_ms: u64) - return resp(sorted_set::zrevrangebyscore_readonly(db, args, now_ms)); } } + (14, b'z') => { + // ZREVRANGEBYLEX + if cmd.eq_ignore_ascii_case(b"ZREVRANGEBYLEX") { + return resp(sorted_set::zrevrangebylex_readonly(db, args, now_ms)); + } + } // ---- new arms (contract v2): buckets that don't conflict with pre-existing ones ---- (3, b'l') => { // LCS @@ -1728,6 +1760,9 @@ fn dispatch_read_inner(db: &Database, cmd: &[u8], args: &[Frame], now_ms: u64) - if cmd.eq_ignore_ascii_case(b"ZRANDMEMBER") { return resp(sorted_set::zrandmember_readonly(db, args, now_ms)); } + if cmd.eq_ignore_ascii_case(b"ZRANGEBYLEX") { + return resp(sorted_set::zrangebylex_readonly(db, args, now_ms)); + } } (11, b'b') => { // BITFIELD_RO (11 bytes) @@ -2360,6 +2395,8 @@ mod tests { b"ZLEXCOUNT", b"ZRANGEBYSCORE", b"ZREVRANGEBYSCORE", + b"ZRANGEBYLEX", + b"ZREVRANGEBYLEX", b"LLEN", b"LRANGE", b"LINDEX", diff --git a/src/command/sorted_set/mod.rs b/src/command/sorted_set/mod.rs index a241ed533..997e80961 100644 --- a/src/command/sorted_set/mod.rs +++ b/src/command/sorted_set/mod.rs @@ -1,8 +1,12 @@ +mod sorted_set_lex; mod sorted_set_read; +mod sorted_set_store; mod sorted_set_write; mod work_budget; +pub use sorted_set_lex::*; pub use sorted_set_read::*; +pub use sorted_set_store::*; pub use sorted_set_write::*; use bytes::Bytes; @@ -371,6 +375,43 @@ pub(super) fn lex_in_range(member: &[u8], min: &LexBound, max: &LexBound) -> boo // Shared range helpers // --------------------------------------------------------------------------- +/// Resolve a `start stop` rank pair the way Redis's `zremrangeGenericCommand` +/// does, returning the inclusive window or `None` when it is empty. +/// +/// A negative index counts from the end. A START still negative after that is +/// clamped to 0; a STOP still negative is NOT, so `start > stop` reports the +/// window empty — which is what makes `ZREMRANGEBYRANK z -10 -6` on a +/// five-member zset remove nothing (redis 8.6.1 answers `(integer) 0`). +/// +/// `zrange_by_rank` and `zrange_from_entries` below clamp the STOP as well and +/// answer `[a]` for the same `ZRANGE z -10 -6`, where redis answers `[]`. That +/// is a pre-existing divergence in a command moon#959 does not touch; it is +/// reported separately rather than changed under this issue, and this helper +/// exists so the new command does not inherit it. +pub(super) fn rank_window(start_raw: i64, stop_raw: i64, total: usize) -> Option<(usize, usize)> { + let len = total as i64; + let mut start = if start_raw < 0 { + len.saturating_add(start_raw) + } else { + start_raw + }; + let mut stop = if stop_raw < 0 { + len.saturating_add(stop_raw) + } else { + stop_raw + }; + if start < 0 { + start = 0; + } + if start > stop || start >= len { + return None; + } + if stop >= len { + stop = len - 1; + } + Some((start as usize, stop as usize)) +} + pub(super) fn zrange_by_rank( scores: &BPTree, min_arg: &[u8], @@ -3984,3 +4025,975 @@ mod zadd_listpack_batch_tests { ); } } + +/// moon#959 — the six commands that used to be `unknown command`, and +/// `ZADD ... INCR`. Every expectation below was read off redis-server 8.6.1 +/// on the wire (`/tmp/z959/oracle_vs_control.txt` in the PR) before the code +/// was written; the reply bytes are what these assert, not `COMMAND INFO`. +/// +/// Dispatch-path coverage, stated per CLAUDE.md's three-path rule: +/// * `command::dispatch` (the mutable path MULTI/EXEC, Lua and every write +/// take) — every `call(...)` below goes through it. +/// * `command::dispatch_read` (the shared-lock path a bare read takes) — +/// `call_read(...)` for the two lex reads, plus the prefilter check in +/// `dispatch_read_serves_the_lex_reads`. +/// * `server::conn::try_inline_dispatch` inlines exactly `GET` and a plain +/// `SET` (`blocking.rs`); every other command falls through to generic +/// dispatch, so there is no arm for a zset command to be missing from. +#[cfg(test)] +mod missing_commands_959_tests { + use super::*; + use crate::command::{DispatchResult, dispatch, dispatch_read, is_dispatch_read_supported}; + use crate::storage::Database; + + fn bs(s: &[u8]) -> Frame { + Frame::BulkString(Bytes::copy_from_slice(s)) + } + + fn argv(args: &[&str]) -> Vec { + args.iter().map(|a| bs(a.as_bytes())).collect() + } + + /// Through the real mutable dispatch table, so a missing arm shows up as + /// `unknown command` rather than as a handler that was never reached. + fn call(db: &mut Database, cmd: &str, args: &[&str]) -> Frame { + let mut selected = 0usize; + match dispatch(db, cmd.as_bytes(), &argv(args), &mut selected, 16) { + DispatchResult::Response(f) => f, + DispatchResult::Quit(f) => panic!("unexpected Quit for {cmd}: {f:?}"), + } + } + + /// Through the shared-lock read table. + fn call_read(db: &Database, cmd: &str, args: &[&str]) -> Frame { + let mut selected = 0usize; + let now_ms = db.now_ms(); + match dispatch_read(db, cmd.as_bytes(), &argv(args), now_ms, &mut selected, 16) { + DispatchResult::Response(f) => f, + DispatchResult::Quit(f) => panic!("unexpected Quit for {cmd}: {f:?}"), + } + } + + fn seed(db: &mut Database, key: &str, pairs: &[(&str, &str)]) { + let mut a = vec![key]; + for (s, m) in pairs { + a.push(s); + a.push(m); + } + let n = call(db, "ZADD", &a); + assert_eq!(n, Frame::Integer(pairs.len() as i64), "seeding {key}"); + } + + const FIVE: &[(&str, &str)] = &[("1", "a"), ("2", "b"), ("3", "c"), ("4", "d"), ("5", "e")]; + const LEX: &[(&str, &str)] = &[("0", "a"), ("0", "b"), ("0", "c"), ("0", "d"), ("0", "e")]; + + /// A member past `zset-max-listpack-value` (64) forces the B+tree form. + /// All `z`s so it sorts AFTER every fixture member under a lex range. + const LONG: &str = "zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz"; + + fn strings(frame: &Frame) -> Vec { + match frame { + Frame::Array(items) => items + .iter() + .map(|f| match f { + Frame::BulkString(b) => String::from_utf8_lossy(b).into_owned(), + other => panic!("not a bulk string: {other:?}"), + }) + .collect(), + other => panic!("not an array: {other:?}"), + } + } + + fn range(db: &mut Database, key: &str) -> Vec { + strings(&call(db, "ZRANGE", &[key, "0", "-1"])) + } + + fn err_text(frame: &Frame) -> String { + match frame { + Frame::Error(e) => String::from_utf8_lossy(e).into_owned(), + other => panic!("expected an error reply, got {other:?}"), + } + } + + fn encoding_of(db: &mut Database, key: &str) -> String { + match crate::command::key::object(db, &[bs(b"ENCODING"), bs(key.as_bytes())]) { + Frame::BulkString(b) => String::from_utf8_lossy(&b).into_owned(), + other => panic!("OBJECT ENCODING did not reply a bulk string: {other:?}"), + } + } + + fn exists(db: &mut Database, key: &str) -> bool { + call(db, "EXISTS", &[key]) == Frame::Integer(1) + } + + fn ledger_exact(db: &mut Database, step: &str) { + let running = db.estimated_memory(); + db.recalculate_memory(); + let recomputed = db.estimated_memory(); + assert_eq!( + running, recomputed, + "{step}: ledger {running} != recount {recomputed}" + ); + } + + // ── the tripwire: nothing here is `unknown command` any more ──────── + + #[test] + fn all_six_are_dispatched_and_the_negative_control_is_not() { + let mut db = Database::new(); + seed(&mut db, "z", FIVE); + for (cmd, args) in [ + ("ZRANGEBYLEX", &["z", "-", "+"][..]), + ("ZREVRANGEBYLEX", &["z", "+", "-"][..]), + ("ZREMRANGEBYRANK", &["z", "0", "0"][..]), + ("ZREMRANGEBYSCORE", &["z", "0", "0"][..]), + ("ZREMRANGEBYLEX", &["z", "[zz", "[zz"][..]), + ("ZDIFFSTORE", &["d", "1", "z"][..]), + ] { + let reply = call(&mut db, cmd, args); + assert!( + !matches!(&reply, Frame::Error(e) if e.starts_with(b"ERR unknown command")), + "{cmd} is still unknown to dispatch: {reply:?}" + ); + // Lower-case, as redis-py sends it. + let reply = call(&mut db, &cmd.to_ascii_lowercase(), args); + assert!( + !matches!(&reply, Frame::Error(e) if e.starts_with(b"ERR unknown command")), + "{cmd} (lower-case) is still unknown to dispatch: {reply:?}" + ); + } + // The negative control: the same shape the issue used, still refused. + let reply = call(&mut db, "ZNOTACOMMAND", &["z"]); + assert!( + err_text(&reply).starts_with("ERR unknown command"), + "{reply:?}" + ); + } + + #[test] + fn registry_carries_the_six_with_the_sortedset_category() { + use crate::command::metadata::{AclCategories, CommandFlags, lookup}; + for (name, write, arity) in [ + ("ZRANGEBYLEX", false, -4), + ("ZREVRANGEBYLEX", false, -4), + ("ZREMRANGEBYRANK", true, 4), + ("ZREMRANGEBYSCORE", true, 4), + ("ZREMRANGEBYLEX", true, 4), + ("ZDIFFSTORE", true, -4), + ] { + let meta = lookup(name.as_bytes()).unwrap_or_else(|| panic!("{name} not registered")); + assert_eq!(meta.arity, arity, "{name} arity"); + assert_eq!( + meta.flags.contains(CommandFlags::WRITE), + write, + "{name} write flag" + ); + assert_eq!( + meta.flags.contains(CommandFlags::READONLY), + !write, + "{name} read flag" + ); + assert!( + meta.acl_categories.contains(AclCategories::SORTEDSET), + "{name} must be @sortedset" + ); + assert_eq!(meta.first_key, 1, "{name} first key"); + } + } + + #[test] + fn dispatch_read_serves_the_lex_reads() { + let mut db = Database::new(); + seed(&mut db, "lex", LEX); + assert!(is_dispatch_read_supported(b"ZRANGEBYLEX")); + assert!(is_dispatch_read_supported(b"ZREVRANGEBYLEX")); + assert_eq!( + strings(&call_read(&db, "ZRANGEBYLEX", &["lex", "[b", "(d"])), + ["b", "c"] + ); + assert_eq!( + strings(&call_read(&db, "ZREVRANGEBYLEX", &["lex", "(d", "[b"])), + ["c", "b"] + ); + // The read path answered from the listpack without flattening it. + assert_eq!(encoding_of(&mut db, "lex"), "listpack"); + } + + // ── ZRANGEBYLEX / ZREVRANGEBYLEX ───────────────────────────────────── + + #[test] + fn zrangebylex_bounds_and_limit_match_the_oracle() { + for promote in [false, true] { + let mut db = Database::new(); + seed(&mut db, "lex", LEX); + if promote { + call(&mut db, "ZADD", &["lex", "0", LONG]); + assert_eq!(encoding_of(&mut db, "lex"), "skiplist"); + } + let r = |db: &mut Database, a: &[&str]| strings(&call(db, "ZRANGEBYLEX", a)); + let tail: &[&str] = if promote { &[LONG] } else { &[] }; + let mut all = vec!["a", "b", "c", "d", "e"]; + all.extend_from_slice(tail); + assert_eq!(r(&mut db, &["lex", "-", "+"]), all, "promote={promote}"); + assert_eq!(r(&mut db, &["lex", "[b", "(d"]), ["b", "c"]); + assert_eq!(r(&mut db, &["lex", "(b", "[d"]), ["c", "d"]); + assert_eq!(r(&mut db, &["lex", "[c", "[c"]), ["c"]); + // `(` and `[` alone are exclusive/inclusive EMPTY strings: every + // member is > "" and none is <= "". + assert!(r(&mut db, &["lex", "(", "["]).is_empty()); + assert_eq!( + r(&mut db, &["lex", "-", "+", "LIMIT", "1", "2"]), + ["b", "c"] + ); + assert!(r(&mut db, &["lex", "-", "+", "LIMIT", "-1", "2"]).is_empty()); + assert!(r(&mut db, &["lex", "-", "+", "LIMIT", "0", "0"]).is_empty()); + let mut from_b = vec!["b", "c", "d", "e"]; + from_b.extend_from_slice(tail); + assert_eq!(r(&mut db, &["lex", "-", "+", "LIMIT", "1", "-1"]), from_b); + // Reversed bounds are an empty range, not an error. + assert!(r(&mut db, &["lex", "+", "-"]).is_empty()); + // Lower-case option token. + assert_eq!( + r(&mut db, &["lex", "-", "+", "limit", "1", "2"]), + ["b", "c"] + ); + assert!(r(&mut db, &["nokey", "-", "+"]).is_empty()); + } + } + + #[test] + fn zrevrangebylex_takes_max_then_min_and_walks_backwards() { + let mut db = Database::new(); + seed(&mut db, "lex", LEX); + let r = |db: &mut Database, a: &[&str]| strings(&call(db, "ZREVRANGEBYLEX", a)); + assert_eq!(r(&mut db, &["lex", "+", "-"]), ["e", "d", "c", "b", "a"]); + assert_eq!(r(&mut db, &["lex", "[d", "(b"]), ["d", "c"]); + assert_eq!(r(&mut db, &["lex", "(d", "[b"]), ["c", "b"]); + assert!(r(&mut db, &["lex", "-", "+"]).is_empty()); + assert_eq!( + r(&mut db, &["lex", "+", "-", "LIMIT", "1", "2"]), + ["d", "c"] + ); + assert_eq!( + r(&mut db, &["lex", "+", "-", "LIMIT", "1", "-1"]), + ["d", "c", "b", "a"] + ); + assert!(r(&mut db, &["lex", "+", "-", "LIMIT", "-1", "2"]).is_empty()); + } + + #[test] + fn zrangebylex_error_surface_matches_the_oracle() { + let mut db = Database::new(); + seed(&mut db, "lex", LEX); + call(&mut db, "SET", &["str", "v"]); + let e = |db: &mut Database, cmd: &str, a: &[&str]| err_text(&call(db, cmd, a)); + for cmd in ["ZRANGEBYLEX", "ZREVRANGEBYLEX"] { + let lc = cmd.to_ascii_lowercase(); + assert_eq!( + e(&mut db, cmd, &["lex", "-"]), + format!("ERR wrong number of arguments for '{lc}' command") + ); + assert_eq!( + e(&mut db, cmd, &["lex", "a", "b"]), + "ERR min or max not valid string range item" + ); + assert_eq!( + e(&mut db, cmd, &["lex", "", "+"]), + "ERR min or max not valid string range item" + ); + // The grammar is checked BEFORE the key: a missing key with a bad + // bound is still an error, not an empty array. + assert_eq!( + e(&mut db, cmd, &["nokey", "a", "b"]), + "ERR min or max not valid string range item" + ); + assert_eq!( + e(&mut db, cmd, &["lex", "-", "+", "WITHSCORES"]), + "ERR syntax error, WITHSCORES not supported in combination with BYLEX" + ); + // ... and the WITHSCORES refusal outranks a bad bound. (The + // first draft had these the other way round; the oracle sweep of + // the built binary caught it, which is why every row is sent.) + assert_eq!( + e(&mut db, cmd, &["lex", "a", "b", "WITHSCORES"]), + "ERR syntax error, WITHSCORES not supported in combination with BYLEX" + ); + assert_eq!( + e(&mut db, cmd, &["lex", "-", "+", "LIMIT", "1"]), + "ERR syntax error" + ); + assert_eq!( + e(&mut db, cmd, &["lex", "-", "+", "BOGUS"]), + "ERR syntax error" + ); + assert_eq!( + e(&mut db, cmd, &["lex", "-", "+", "LIMIT", "notanint", "1"]), + "ERR value is not an integer or out of range" + ); + assert_eq!( + e(&mut db, cmd, &["lex", "-", "+", "LIMIT", "1", "notanint"]), + "ERR value is not an integer or out of range" + ); + // The option loop runs first: a dangling LIMIT beats a bad bound. + assert_eq!( + e(&mut db, cmd, &["lex", "a", "b", "LIMIT", "1"]), + "ERR syntax error" + ); + assert!(e(&mut db, cmd, &["str", "-", "+"]).starts_with("WRONGTYPE")); + } + } + + // ── ZREMRANGEBYRANK ────────────────────────────────────────────────── + + #[test] + fn zremrangebyrank_normalises_ranks_like_redis() { + for promote in [false, true] { + let cases: &[(&str, &str, i64, &[&str])] = &[ + ("0", "0", 1, &["b", "c", "d", "e"]), + ("-2", "-1", 2, &["a", "b", "c"]), + ("3", "1", 0, &["a", "b", "c", "d", "e"]), + ("0", "100", 5, &[]), + ("-100", "1", 2, &["c", "d", "e"]), + ("5", "10", 0, &["a", "b", "c", "d", "e"]), + ("-1", "-3", 0, &["a", "b", "c", "d", "e"]), + // A stop still negative after normalisation is NOT clamped + // to 0: redis 8.6.1 removes nothing here. + ("-10", "-6", 0, &["a", "b", "c", "d", "e"]), + ("2", "-2", 2, &["a", "b", "e"]), + ("0", "-1", 5, &[]), + ]; + for (start, stop, removed, left) in cases { + let mut db = Database::new(); + seed(&mut db, "r", FIVE); + if promote { + // Promote WITHOUT changing the membership under test. + call(&mut db, "ZADD", &["r", "9", LONG]); + call(&mut db, "ZREM", &["r", LONG]); + assert_eq!(encoding_of(&mut db, "r"), "skiplist"); + } else { + assert_eq!(encoding_of(&mut db, "r"), "listpack"); + } + assert_eq!( + call(&mut db, "ZREMRANGEBYRANK", &["r", start, stop]), + Frame::Integer(*removed), + "ZREMRANGEBYRANK r {start} {stop} promote={promote}" + ); + if left.is_empty() { + assert!(!exists(&mut db, "r"), "drained key must be gone"); + } else { + assert_eq!( + range(&mut db, "r"), + *left, + "{start} {stop} promote={promote}" + ); + // A removal never converts the encoding (moon#897). + assert_eq!( + encoding_of(&mut db, "r"), + if promote { "skiplist" } else { "listpack" } + ); + } + ledger_exact( + &mut db, + &format!("ZREMRANGEBYRANK {start} {stop} promote={promote}"), + ); + } + } + } + + #[test] + fn zremrangebyrank_error_surface_and_missing_key() { + let mut db = Database::new(); + seed(&mut db, "r", FIVE); + call(&mut db, "SET", &["str", "v"]); + let e = |db: &mut Database, a: &[&str]| err_text(&call(db, "ZREMRANGEBYRANK", a)); + for bad in [ + &["r", "notanint", "1"][..], + &["r", "1", "notanint"], + &["r", "1.5", "2"], + ] { + assert_eq!( + e(&mut db, bad), + "ERR value is not an integer or out of range" + ); + } + for bad in [&["r", "1"][..], &["r", "1", "2", "3"], &["r"]] { + assert_eq!( + e(&mut db, bad), + "ERR wrong number of arguments for 'zremrangebyrank' command" + ); + } + assert!(e(&mut db, &["str", "0", "1"]).starts_with("WRONGTYPE")); + assert_eq!( + range(&mut db, "r").len(), + 5, + "no error may have removed anything" + ); + // A bad index on a MISSING key is still the integer error (the range + // is parsed before the lookup), and a good one answers 0 and creates + // nothing. + assert_eq!( + e(&mut db, &["nokey", "x", "1"]), + "ERR value is not an integer or out of range" + ); + assert_eq!( + call(&mut db, "ZREMRANGEBYRANK", &["nokey", "0", "1"]), + Frame::Integer(0) + ); + assert!(!exists(&mut db, "nokey")); + ledger_exact(&mut db, "after the error surface"); + } + + // ── ZREMRANGEBYSCORE ───────────────────────────────────────────────── + + #[test] + fn zremrangebyscore_bounds_match_the_oracle() { + for promote in [false, true] { + let cases: &[(&str, &str, i64, &[&str])] = &[ + ("2", "3", 2, &["a", "d", "e"]), + ("(2", "3", 1, &["a", "b", "d", "e"]), + ("-inf", "+inf", 5, &[]), + ("3", "1", 0, &["a", "b", "c", "d", "e"]), + ("+inf", "-inf", 0, &["a", "b", "c", "d", "e"]), + ("(1", "(1", 0, &["a", "b", "c", "d", "e"]), + ("(1", "2", 1, &["a", "c", "d", "e"]), + ("(5", "inf", 0, &["a", "b", "c", "d", "e"]), + ("5", "inf", 1, &["a", "b", "c", "d"]), + ]; + for (min, max, removed, left) in cases { + let mut db = Database::new(); + seed(&mut db, "s", FIVE); + if promote { + call(&mut db, "ZADD", &["s", "9", LONG]); + call(&mut db, "ZREM", &["s", LONG]); + assert_eq!(encoding_of(&mut db, "s"), "skiplist"); + } + assert_eq!( + call(&mut db, "ZREMRANGEBYSCORE", &["s", min, max]), + Frame::Integer(*removed), + "ZREMRANGEBYSCORE s {min} {max} promote={promote}" + ); + if left.is_empty() { + assert!(!exists(&mut db, "s")); + } else { + assert_eq!(range(&mut db, "s"), *left, "{min} {max} promote={promote}"); + } + ledger_exact( + &mut db, + &format!("ZREMRANGEBYSCORE {min} {max} promote={promote}"), + ); + } + } + } + + #[test] + fn zremrangebyscore_error_surface_and_missing_key() { + let mut db = Database::new(); + seed(&mut db, "s", FIVE); + call(&mut db, "SET", &["str", "v"]); + let e = |db: &mut Database, a: &[&str]| err_text(&call(db, "ZREMRANGEBYSCORE", a)); + for bad in [&["s", "nan", "1"][..], &["s", "a", "1"], &["s", "1", "a"]] { + assert_eq!(e(&mut db, bad), "ERR min or max is not a float"); + } + for bad in [&["s", "1"][..], &["s", "1", "2", "3"]] { + assert_eq!( + e(&mut db, bad), + "ERR wrong number of arguments for 'zremrangebyscore' command" + ); + } + assert!(e(&mut db, &["str", "0", "1"]).starts_with("WRONGTYPE")); + assert_eq!( + e(&mut db, &["nokey", "x", "1"]), + "ERR min or max is not a float" + ); + assert_eq!( + call(&mut db, "ZREMRANGEBYSCORE", &["nokey", "0", "1"]), + Frame::Integer(0) + ); + assert!(!exists(&mut db, "nokey")); + assert_eq!(range(&mut db, "s").len(), 5); + } + + // ── ZREMRANGEBYLEX ─────────────────────────────────────────────────── + + #[test] + fn zremrangebylex_bounds_match_the_oracle() { + for promote in [false, true] { + let cases: &[(&str, &str, i64, &[&str])] = &[ + ("[b", "(d", 2, &["a", "d", "e"]), + ("-", "+", 5, &[]), + ("+", "-", 0, &["a", "b", "c", "d", "e"]), + ("(c", "+", 2, &["a", "b", "c"]), + ("[zz", "[zz", 0, &["a", "b", "c", "d", "e"]), + ]; + for (min, max, removed, left) in cases { + let mut db = Database::new(); + seed(&mut db, "l", LEX); + if promote { + call(&mut db, "ZADD", &["l", "0", LONG]); + call(&mut db, "ZREM", &["l", LONG]); + assert_eq!(encoding_of(&mut db, "l"), "skiplist"); + } + assert_eq!( + call(&mut db, "ZREMRANGEBYLEX", &["l", min, max]), + Frame::Integer(*removed), + "ZREMRANGEBYLEX l {min} {max} promote={promote}" + ); + if left.is_empty() { + assert!(!exists(&mut db, "l")); + } else { + assert_eq!(range(&mut db, "l"), *left, "{min} {max} promote={promote}"); + } + ledger_exact( + &mut db, + &format!("ZREMRANGEBYLEX {min} {max} promote={promote}"), + ); + } + } + } + + #[test] + fn zremrangebylex_error_surface_and_missing_key() { + let mut db = Database::new(); + seed(&mut db, "l", LEX); + call(&mut db, "SET", &["str", "v"]); + let e = |db: &mut Database, a: &[&str]| err_text(&call(db, "ZREMRANGEBYLEX", a)); + assert_eq!( + e(&mut db, &["l", "a", "b"]), + "ERR min or max not valid string range item" + ); + assert_eq!( + e(&mut db, &["l", "", "+"]), + "ERR min or max not valid string range item" + ); + for bad in [&["l", "-"][..], &["l", "-", "+", "x"]] { + assert_eq!( + e(&mut db, bad), + "ERR wrong number of arguments for 'zremrangebylex' command" + ); + } + assert!(e(&mut db, &["str", "-", "+"]).starts_with("WRONGTYPE")); + assert_eq!( + e(&mut db, &["nokey", "x", "1"]), + "ERR min or max not valid string range item" + ); + assert_eq!( + call(&mut db, "ZREMRANGEBYLEX", &["nokey", "-", "+"]), + Frame::Integer(0) + ); + assert!(!exists(&mut db, "nokey")); + assert_eq!(range(&mut db, "l").len(), 5); + } + + // ── ZDIFFSTORE ─────────────────────────────────────────────────────── + + fn scored(db: &mut Database, key: &str) -> Vec { + strings(&call(db, "ZRANGE", &[key, "0", "-1", "WITHSCORES"])) + } + + #[test] + fn zdiffstore_computes_the_difference_with_first_source_scores() { + let mut db = Database::new(); + seed(&mut db, "z", FIVE); + seed(&mut db, "z2", &[("1", "a"), ("2", "b")]); + seed(&mut db, "z3", &[("2", "b"), ("9", "x")]); + assert_eq!( + call(&mut db, "ZDIFFSTORE", &["d1", "2", "z", "z2"]), + Frame::Integer(3) + ); + assert_eq!(scored(&mut db, "d1"), ["c", "3", "d", "4", "e", "5"]); + assert_eq!( + call(&mut db, "ZDIFFSTORE", &["d2", "1", "z"]), + Frame::Integer(5) + ); + assert_eq!( + call(&mut db, "ZDIFFSTORE", &["d3", "2", "z", "nokey"]), + Frame::Integer(5) + ); + assert_eq!( + call(&mut db, "ZDIFFSTORE", &["d8", "3", "z", "z2", "z3"]), + Frame::Integer(3) + ); + assert_eq!(scored(&mut db, "d8"), ["c", "3", "d", "4", "e", "5"]); + // Sources are read before the destination is replaced, so a + // destination that is also a source is diffed from its OLD content. + assert_eq!( + call(&mut db, "ZDIFFSTORE", &["z3", "2", "z", "z3"]), + Frame::Integer(4) + ); + assert_eq!( + scored(&mut db, "z3"), + ["a", "1", "c", "3", "d", "4", "e", "5"] + ); + assert_eq!( + call(&mut db, "ZDIFFSTORE", &["z2", "1", "z2"]), + Frame::Integer(2) + ); + assert_eq!(scored(&mut db, "z2"), ["a", "1", "b", "2"]); + // Lower-case, as a client library sends it. + assert_eq!( + call(&mut db, "zdiffstore", &["d9", "1", "z"]), + Frame::Integer(5) + ); + // Reading a listpack source did not flatten it. + assert_eq!(encoding_of(&mut db, "z"), "listpack"); + ledger_exact(&mut db, "after the ZDIFFSTORE happy paths"); + } + + #[test] + fn zdiffstore_empty_result_deletes_the_destination() { + let mut db = Database::new(); + seed(&mut db, "z", FIVE); + assert_eq!( + call(&mut db, "ZDIFFSTORE", &["d4", "2", "nokey", "z"]), + Frame::Integer(0) + ); + assert!(!exists(&mut db, "d4")); + // Even a destination of another type is replaced — by nothing. + call(&mut db, "SET", &["d5", "x"]); + assert_eq!( + call(&mut db, "ZDIFFSTORE", &["d5", "2", "z", "z"]), + Frame::Integer(0) + ); + assert!(!exists(&mut db, "d5")); + ledger_exact(&mut db, "after an empty ZDIFFSTORE"); + } + + #[test] + fn zdiffstore_error_surface_matches_the_oracle() { + let mut db = Database::new(); + seed(&mut db, "z", FIVE); + call(&mut db, "SET", &["str", "v"]); + let e = |db: &mut Database, a: &[&str]| err_text(&call(db, "ZDIFFSTORE", a)); + // The two-class numkeys split (moon#969). + assert_eq!( + e(&mut db, &["d", "0", "z"]), + "ERR at least 1 input key is needed for 'zdiffstore' command" + ); + assert_eq!( + e(&mut db, &["d", "-1", "z"]), + "ERR at least 1 input key is needed for 'zdiffstore' command" + ); + assert_eq!( + e(&mut db, &["d", "notanint", "z"]), + "ERR value is not an integer or out of range" + ); + // Arity first: no key named at all. + for bad in [&["d", "1"][..], &["d"], &["d", "0"]] { + assert_eq!( + e(&mut db, bad), + "ERR wrong number of arguments for 'zdiffstore' command" + ); + } + // numkeys overrunning the key list, and every option token: ZDIFFSTORE + // takes none, so WEIGHTS/AGGREGATE are as unknown as BOGUS. + assert_eq!(e(&mut db, &["d", "2", "z"]), "ERR syntax error"); + for opts in [ + &["WEIGHTS", "1"][..], + &["AGGREGATE", "SUM"], + &["WITHSCORES"], + &["BOGUS"], + ] { + let mut a = vec!["d", "1", "z"]; + a.extend_from_slice(opts); + assert_eq!(e(&mut db, &a), "ERR syntax error", "{opts:?}"); + } + assert!( + !exists(&mut db, "d"), + "no error may have created the destination" + ); + // WRONGTYPE from either position, and it outranks an option error: + // Redis looks the sources up before it parses the options. + assert!(e(&mut db, &["d", "2", "str", "z"]).starts_with("WRONGTYPE")); + assert!(e(&mut db, &["d", "2", "z", "str"]).starts_with("WRONGTYPE")); + assert!(e(&mut db, &["d", "1", "str", "BOGUS"]).starts_with("WRONGTYPE")); + // ... while a numkeys error or an overrun is decided before the lookup. + assert_eq!( + e(&mut db, &["d", "0", "str", "BOGUS"]), + "ERR at least 1 input key is needed for 'zdiffstore' command" + ); + assert_eq!(e(&mut db, &["d", "2", "str"]), "ERR syntax error"); + // The destination's type is irrelevant until the write. + assert_eq!(e(&mut db, &["str", "1", "z", "BOGUS"]), "ERR syntax error"); + assert!(!exists(&mut db, "d")); + } + + /// The precedence fix above applies to the whole family, since the three + /// share one implementation: `ZUNIONSTORE d 1 BOGUS` is + /// WRONGTYPE on redis 8.6.1, and was `syntax error` on moon. + #[test] + fn zunionstore_wrongtype_outranks_an_option_error() { + let mut db = Database::new(); + seed(&mut db, "z", FIVE); + call(&mut db, "SET", &["str", "v"]); + for (cmd, opts) in [ + ("ZUNIONSTORE", &["BOGUS"][..]), + ("ZUNIONSTORE", &["WEIGHTS", "nan"]), + ("ZINTERSTORE", &["WEIGHTS", "1"]), + ] { + let mut a = vec!["d", "2", "z", "str"]; + a.extend_from_slice(opts); + assert!( + err_text(&call(&mut db, cmd, &a)).starts_with("WRONGTYPE"), + "{cmd} {opts:?}" + ); + } + // A well-typed source with a bad option is still the option's error. + assert_eq!( + err_text(&call(&mut db, "ZUNIONSTORE", &["d", "1", "z", "BOGUS"])), + "ERR syntax error" + ); + assert_eq!( + err_text(&call( + &mut db, + "ZUNIONSTORE", + &["d", "1", "z", "WEIGHTS", "nan"] + )), + "ERR weight value is not a float" + ); + // And reading a listpack source through the store family leaves it a + // listpack (the moon#928 defect, closed for this family too). + assert_eq!( + call(&mut db, "ZUNIONSTORE", &["u", "1", "z"]), + Frame::Integer(5) + ); + assert_eq!(encoding_of(&mut db, "z"), "listpack"); + } + + // ── ZADD ... INCR ──────────────────────────────────────────────────── + + fn bulk_text(frame: &Frame) -> String { + match frame { + Frame::BulkString(b) => String::from_utf8_lossy(b).into_owned(), + other => panic!("expected a bulk string, got {other:?}"), + } + } + + #[test] + fn zadd_incr_replies_the_new_score_on_both_encodings() { + for promote in [false, true] { + let mut db = Database::new(); + let key = "i"; + if promote { + call(&mut db, "ZADD", &[key, "1", LONG]); + assert_eq!(encoding_of(&mut db, key), "skiplist"); + } + let incr = |db: &mut Database, a: &[&str]| call(db, "ZADD", a); + assert_eq!(bulk_text(&incr(&mut db, &[key, "INCR", "5", "a"])), "5"); + assert_eq!(bulk_text(&incr(&mut db, &[key, "INCR", "2.5", "a"])), "7.5"); + assert_eq!( + bulk_text(&incr(&mut db, &[key, "INCR", "1e3", "big"])), + "1000" + ); + assert_eq!( + bulk_text(&incr(&mut db, &[key, "INCR", "0.1", "big"])), + "1000.1" + ); + // Option order does not matter, and CH has no say in the reply. + assert_eq!( + bulk_text(&incr(&mut db, &[key, "CH", "INCR", "1", "a"])), + "8.5" + ); + assert_eq!( + bulk_text(&incr(&mut db, &[key, "INCR", "CH", "1", "a"])), + "9.5" + ); + assert_eq!(bulk_text(&incr(&mut db, &[key, "incr", "1", "a"])), "10.5"); + assert_eq!(bulk_text(&call(&mut db, "ZSCORE", &[key, "a"])), "10.5"); + if !promote { + assert_eq!(encoding_of(&mut db, key), "listpack"); + } + ledger_exact(&mut db, &format!("after ZADD INCR promote={promote}")); + } + } + + #[test] + fn zadd_incr_honours_nx_xx_gt_lt_like_redis() { + for promote in [false, true] { + let mut db = Database::new(); + let key = "i"; + if promote { + call(&mut db, "ZADD", &[key, "1", LONG]); + } + let incr = |db: &mut Database, a: &[&str]| call(db, "ZADD", a); + assert_eq!(bulk_text(&incr(&mut db, &[key, "INCR", "5", "a"])), "5"); + // NX: refuses a present member, admits a new one. + assert_eq!(incr(&mut db, &[key, "NX", "INCR", "1", "a"]), Frame::Null); + assert_eq!( + bulk_text(&incr(&mut db, &[key, "NX", "INCR", "1", "newm"])), + "1" + ); + // XX: refuses a new member, admits a present one. + assert_eq!( + incr(&mut db, &[key, "XX", "INCR", "1", "nope"]), + Frame::Null + ); + assert_eq!( + bulk_text(&incr(&mut db, &[key, "XX", "INCR", "1", "a"])), + "6" + ); + // GT/LT: only a move in the right direction; zero is a refusal. + assert_eq!(incr(&mut db, &[key, "GT", "INCR", "-1", "a"]), Frame::Null); + assert_eq!( + bulk_text(&incr(&mut db, &[key, "GT", "INCR", "1", "a"])), + "7" + ); + assert_eq!(incr(&mut db, &[key, "LT", "INCR", "1", "a"]), Frame::Null); + assert_eq!( + bulk_text(&incr(&mut db, &[key, "LT", "INCR", "-1", "a"])), + "6" + ); + assert_eq!(incr(&mut db, &[key, "GT", "INCR", "0", "a"]), Frame::Null); + assert_eq!(incr(&mut db, &[key, "LT", "INCR", "0", "a"]), Frame::Null); + // GT/LT never block a first insert; XX+GT does. + assert_eq!( + bulk_text(&incr(&mut db, &[key, "GT", "INCR", "1", "zz"])), + "1" + ); + assert_eq!( + bulk_text(&incr(&mut db, &[key, "LT", "INCR", "1", "yy"])), + "1" + ); + assert_eq!( + incr(&mut db, &[key, "XX", "GT", "INCR", "1", "qq"]), + Frame::Null + ); + // A refusal wrote nothing. + assert_eq!(bulk_text(&call(&mut db, "ZSCORE", &[key, "a"])), "6"); + assert_eq!(call(&mut db, "ZSCORE", &[key, "qq"]), Frame::Null); + ledger_exact( + &mut db, + &format!("after flagged ZADD INCR promote={promote}"), + ); + } + } + + #[test] + fn zadd_incr_refusal_on_a_missing_key_creates_nothing() { + let mut db = Database::new(); + assert_eq!( + call(&mut db, "ZADD", &["i3", "XX", "INCR", "1", "a"]), + Frame::Null + ); + assert!(!exists(&mut db, "i3")); + // The B+tree arm too: a member too long for a listpack. + assert_eq!( + call(&mut db, "ZADD", &["i4", "XX", "INCR", "1", LONG]), + Frame::Null + ); + assert!(!exists(&mut db, "i4")); + assert_eq!( + bulk_text(&call(&mut db, "ZADD", &["i3", "NX", "INCR", "1", "a"])), + "1" + ); + ledger_exact(&mut db, "after refused ZADD INCR on missing keys"); + } + + #[test] + fn zadd_incr_error_surface_matches_the_oracle() { + let mut db = Database::new(); + seed(&mut db, "i", &[("1", "a")]); + call(&mut db, "SET", &["str", "v"]); + let e = |db: &mut Database, a: &[&str]| err_text(&call(db, "ZADD", a)); + assert_eq!( + e(&mut db, &["i", "INCR", "1", "a", "2", "b"]), + "ERR INCR option supports a single increment-element pair" + ); + assert_eq!( + e(&mut db, &["i", "INCR"]), + "ERR wrong number of arguments for 'zadd' command" + ); + // Parity is checked before the pair count ... + assert_eq!(e(&mut db, &["i", "INCR", "1"]), "ERR syntax error"); + assert_eq!( + e(&mut db, &["i", "INCR", "1", "a", "2"]), + "ERR syntax error" + ); + // ... and the flag pairings before both. + assert_eq!( + e(&mut db, &["i", "INCR", "NX", "XX", "1", "a", "2", "b"]), + "ERR XX and NX options at the same time are not compatible" + ); + assert_eq!( + e(&mut db, &["i", "INCR", "GT", "LT", "1", "a"]), + "ERR GT, LT, and/or NX options at the same time are not compatible" + ); + assert_eq!( + e(&mut db, &["i", "INCR", "nan", "a"]), + "ERR value is not a valid float" + ); + assert_eq!( + e(&mut db, &["i", "INCR", "notafloat", "a"]), + "ERR value is not a valid float" + ); + assert!(e(&mut db, &["str", "INCR", "1", "a"]).starts_with("WRONGTYPE")); + // inf + -inf is NaN: refused with the ZINCRBY message, score untouched. + assert_eq!( + bulk_text(&call(&mut db, "ZADD", &["i", "INCR", "inf", "a"])), + "inf" + ); + assert_eq!( + e(&mut db, &["i", "INCR", "-inf", "a"]), + "ERR resulting score is not a number (NaN)" + ); + assert_eq!(bulk_text(&call(&mut db, "ZSCORE", &["i", "a"])), "inf"); + assert_eq!( + encoding_of(&mut db, "i"), + "listpack", + "an erroring INCR must not flatten" + ); + // NX outranks the NaN check: the sum is never formed for a present + // member under NX. + assert_eq!( + call(&mut db, "ZADD", &["i", "NX", "INCR", "-inf", "a"]), + Frame::Null + ); + } + + /// The plain ZINCRBY went through the refactored core; its contract is + /// unchanged. + #[test] + fn zincrby_is_unchanged_by_the_shared_core() { + let mut db = Database::new(); + assert_eq!(bulk_text(&call(&mut db, "ZINCRBY", &["z", "5", "a"])), "5"); + assert_eq!( + bulk_text(&call(&mut db, "ZINCRBY", &["z", "-2.5", "a"])), + "2.5" + ); + assert_eq!( + bulk_text(&call(&mut db, "ZINCRBY", &["z", "inf", "a"])), + "inf" + ); + assert_eq!( + err_text(&call(&mut db, "ZINCRBY", &["z", "-inf", "a"])), + "ERR resulting score is not a number (NaN)" + ); + assert_eq!( + err_text(&call(&mut db, "ZINCRBY", &["z", "nan", "a"])), + "ERR value is not a valid float" + ); + assert_eq!(encoding_of(&mut db, "z"), "listpack"); + assert_eq!(bulk_text(&call(&mut db, "ZINCRBY", &["z", "1", LONG])), "1"); + assert_eq!(encoding_of(&mut db, "z"), "skiplist"); + assert_eq!( + bulk_text(&call(&mut db, "ZINCRBY", &["z", "1", "a"])), + "inf" + ); + ledger_exact(&mut db, "after ZINCRBY through the shared core"); + } + + #[test] + fn rank_window_follows_the_redis_rule() { + assert_eq!(rank_window(0, 0, 5), Some((0, 0))); + assert_eq!(rank_window(-2, -1, 5), Some((3, 4))); + assert_eq!(rank_window(3, 1, 5), None); + assert_eq!(rank_window(0, 100, 5), Some((0, 4))); + assert_eq!(rank_window(-100, 1, 5), Some((0, 1))); + assert_eq!(rank_window(5, 10, 5), None); + assert_eq!(rank_window(-1, -3, 5), None); + assert_eq!(rank_window(-10, -6, 5), None); + assert_eq!(rank_window(2, -2, 5), Some((2, 3))); + assert_eq!(rank_window(0, -1, 0), None); + assert_eq!(rank_window(i64::MIN, i64::MAX, 5), Some((0, 4))); + assert_eq!(rank_window(i64::MAX, i64::MAX, 5), None); + } +} diff --git a/src/command/sorted_set/sorted_set_lex.rs b/src/command/sorted_set/sorted_set_lex.rs new file mode 100644 index 000000000..9764db269 --- /dev/null +++ b/src/command/sorted_set/sorted_set_lex.rs @@ -0,0 +1,160 @@ +//! `ZRANGEBYLEX` and `ZREVRANGEBYLEX` (moon#959). +//! +//! Own file rather than `sorted_set_read.rs`, which already sits at the +//! 1500-line rule. Same shape as `zrangebyscore_readonly`: the mutable-path +//! entry delegates to the shared-borrow twin so a listpack zset survives the +//! read (moon#928), and both encodings answer through the ONE pair of range +//! helpers `ZRANGE ... BYLEX` already uses, so the legacy spelling and the +//! unified one cannot drift apart. Tests stay in `mod.rs`. + +use crate::framevec; +use crate::protocol::Frame; +use crate::storage::Database; + +use crate::command::helpers::{err, err_wrong_args, extract_bytes}; + +use super::{parse_lex_bound, zrange_by_lex, zrange_from_entries}; + +/// ZRANGEBYLEX key min max [LIMIT offset count]. +/// +/// Reads through the shared-borrow implementation so the zset's compact +/// encoding survives the read (moon#928). +pub fn zrangebylex(db: &mut Database, args: &[Frame]) -> Frame { + let now_ms = db.now_ms(); + zrangebylex_readonly(db, args, now_ms) +} + +/// ZREVRANGEBYLEX key max min [LIMIT offset count]. +/// +/// Reads through the shared-borrow implementation so the zset's compact +/// encoding survives the read (moon#928). +pub fn zrevrangebylex(db: &mut Database, args: &[Frame]) -> Frame { + let now_ms = db.now_ms(); + zrevrangebylex_readonly(db, args, now_ms) +} + +/// ZRANGEBYLEX (read-only). +pub fn zrangebylex_readonly(db: &Database, args: &[Frame], now_ms: u64) -> Frame { + zrangebylex_impl(db, args, now_ms, false) +} + +/// ZREVRANGEBYLEX (read-only). +pub fn zrevrangebylex_readonly(db: &Database, args: &[Frame], now_ms: u64) -> Frame { + zrangebylex_impl(db, args, now_ms, true) +} + +/// The one implementation behind both spellings. +/// +/// Error precedence follows Redis's `zrangeGenericCommand`, verified against +/// redis-server 8.6.1: the option loop first (a dangling `LIMIT` or an unknown +/// token is `syntax error`, a non-integer `LIMIT` value is the generic integer +/// error), then `WITHSCORES` — which the legacy spelling parses but refuses +/// with its own message, BEFORE it looks at the bounds — then the range +/// grammar (`min or max not valid string range item`), and only then the key. +/// The bounds are therefore validated BEFORE the lookup, so `ZRANGEBYLEX +/// nokey a b` is an error and not an empty array. The first draft checked the +/// bounds before WITHSCORES; the oracle sweep of the built binary caught it — +/// `ZRANGEBYLEX k a b WITHSCORES` is the WITHSCORES error on redis. +fn zrangebylex_impl(db: &Database, args: &[Frame], now_ms: u64, rev: bool) -> Frame { + let cmd = if rev { "ZREVRANGEBYLEX" } else { "ZRANGEBYLEX" }; + if args.len() < 3 { + return err_wrong_args(cmd); + } + let key = match extract_bytes(&args[0]) { + Some(k) => k, + None => return err_wrong_args(cmd), + }; + // ZREVRANGEBYLEX takes `max min`; the range helpers take `(min, max)` in + // semantic order and only ever use `rev` for iteration direction, exactly + // as `zrevrangebyscore_readonly` does. + let (min_idx, max_idx) = if rev { (2, 1) } else { (1, 2) }; + let min_arg = match extract_bytes(&args[min_idx]) { + Some(b) => b, + None => return err_wrong_args(cmd), + }; + let max_arg = match extract_bytes(&args[max_idx]) { + Some(b) => b, + None => return err_wrong_args(cmd), + }; + + let mut withscores = false; + let mut limit_offset: Option = None; + let mut limit_count: Option = None; + let mut i = 3; + while i < args.len() { + let opt = match extract_bytes(&args[i]) { + Some(b) => b.as_ref(), + None => return err("ERR syntax error"), + }; + if opt.eq_ignore_ascii_case(b"LIMIT") { + // A `LIMIT` with fewer than two values is `syntax error`, not an + // arity error: the arity floor was already met above. + if i + 2 >= args.len() { + return err("ERR syntax error"); + } + let (Some(off_b), Some(cnt_b)) = + (extract_bytes(&args[i + 1]), extract_bytes(&args[i + 2])) + else { + return err("ERR syntax error"); + }; + limit_offset = std::str::from_utf8(off_b).ok().and_then(|s| s.parse().ok()); + limit_count = std::str::from_utf8(cnt_b).ok().and_then(|s| s.parse().ok()); + if limit_offset.is_none() || limit_count.is_none() { + return err("ERR value is not an integer or out of range"); + } + i += 3; + } else if opt.eq_ignore_ascii_case(b"WITHSCORES") { + // Parsed here, refused below, after the whole option loop: a + // later dangling `LIMIT` still wins. + withscores = true; + i += 1; + } else { + return err("ERR syntax error"); + } + } + + if withscores { + return err("ERR syntax error, WITHSCORES not supported in combination with BYLEX"); + } + // Validate the grammar before the key is consulted. The helpers below + // parse the bounds again; that second pass is two small copies on a path + // that is about to materialise the reply, and it keeps the ONE grammar + // `ZRANGE ... BYLEX` uses rather than a second parser to drift from it. + if let Err(e) = parse_lex_bound(min_arg) { + return e; + } + if let Err(e) = parse_lex_bound(max_arg) { + return e; + } + + match db.get_sorted_set_ref_if_alive(key, now_ms) { + Ok(Some(zref)) => match (zref.members_map(), zref.bptree()) { + (Some(members), Some(scores)) => zrange_by_lex( + scores, + min_arg, + max_arg, + rev, + false, + members, + limit_offset, + limit_count, + ), + _ => { + let entries = zref.entries_sorted(); + zrange_from_entries( + &entries, + min_arg, + max_arg, + false, + true, + rev, + false, + limit_offset, + limit_count, + ) + } + }, + Ok(None) => Frame::Array(framevec![]), + Err(e) => e, + } +} diff --git a/src/command/sorted_set/sorted_set_store.rs b/src/command/sorted_set/sorted_set_store.rs new file mode 100644 index 000000000..6d4105b22 --- /dev/null +++ b/src/command/sorted_set/sorted_set_store.rs @@ -0,0 +1,465 @@ +//! The sorted-set STORE family: `ZUNIONSTORE`, `ZINTERSTORE`, `ZDIFFSTORE` +//! and `ZRANGESTORE`. +//! +//! Split out of `sorted_set_write.rs` when `ZDIFFSTORE` and the +//! `ZREMRANGEBY*` trio (moon#959) took that file past the 1500-line rule. +//! Every command here reads one or more SOURCE zsets and REPLACES a +//! destination, which is a different shape from the in-place writes that +//! stay in the write half. Tests stay in `mod.rs`. + +use bytes::Bytes; +use std::collections::HashMap; + +use crate::protocol::Frame; +use crate::storage::Database; +use crate::storage::db::{zset_member_cost, zset_table_bytes}; + +use crate::command::helpers::{err, err_wrong_args, extract_bytes}; + +use super::{ + AggregateOp, clamp_nan_to_zero, parse_numkeys, zadd_member, zrange_by_lex, zrange_by_rank, + zrange_by_score, +}; + +/// Which set operation a `Z*STORE` command computes over its sources. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum SetOp { + Union, + Inter, + /// Members of the FIRST source that are absent from every other one, + /// keeping the first source's scores. Takes no `WEIGHTS`/`AGGREGATE`: + /// Redis's `zunionInterDiffGenericCommand` recognises those tokens only + /// when `op != SET_OP_DIFF`, so on `ZDIFFSTORE` they are `ERR syntax + /// error` like any other unknown token (verified against redis-server + /// 8.6.1, moon#959). + Diff, +} + +impl SetOp { + /// The registered command name, for the arity and `numkeys` messages. + fn name(self) -> &'static str { + match self { + SetOp::Union => "ZUNIONSTORE", + SetOp::Inter => "ZINTERSTORE", + SetOp::Diff => "ZDIFFSTORE", + } + } +} + +/// ZUNIONSTORE destination numkeys key [key ...] [WEIGHTS weight ...] [AGGREGATE SUM|MIN|MAX] +pub fn zunionstore(db: &mut Database, args: &[Frame]) -> Frame { + zstore_impl(db, args, SetOp::Union) +} + +/// ZINTERSTORE destination numkeys key [key ...] [WEIGHTS weight ...] [AGGREGATE SUM|MIN|MAX] +pub fn zinterstore(db: &mut Database, args: &[Frame]) -> Frame { + zstore_impl(db, args, SetOp::Inter) +} + +/// ZDIFFSTORE destination numkeys key [key ...] (moon#959) +/// +/// Stores in `destination` the members of the first source absent from every +/// other source, with the first source's scores. Replies the cardinality of +/// `destination`, deleting it when the difference is empty. Shares the +/// `numkeys` contract and the option loop of its siblings — including the +/// two-class `numkeys` split (moon#969) — with `WEIGHTS`/`AGGREGATE` refused +/// as `syntax error`, which is what redis 8.6.1 answers. +pub fn zdiffstore(db: &mut Database, args: &[Frame]) -> Frame { + zstore_impl(db, args, SetOp::Diff) +} + +fn zstore_impl(db: &mut Database, args: &[Frame], op: SetOp) -> Frame { + let cmd_name = op.name(); + if args.len() < 3 { + return err_wrong_args(cmd_name); + } + let dest = match extract_bytes(&args[0]) { + Some(k) => k, + None => return err_wrong_args(cmd_name), + }; + let numkeys_bytes = match extract_bytes(&args[1]) { + Some(b) => b, + None => return err_wrong_args(cmd_name), + }; + let numkeys = match parse_numkeys(numkeys_bytes, cmd_name) { + Ok(n) => n, + Err(e) => return e, + }; + + // A `numkeys` that overruns the key list is `syntax error`, not an arity + // error (moon#969) — Redis's arity check already passed above, and + // `zunionInterDiffGenericCommand` answers `shared.syntaxerr` here. + if args.len() < 2 + numkeys { + return err("ERR syntax error"); + } + + // Collect source keys + let source_keys: Vec = (0..numkeys) + .map(|j| { + extract_bytes(&args[2 + j]) + .cloned() + .unwrap_or_else(|| Bytes::new()) + }) + .collect(); + + // The sources are read BEFORE the options are parsed, because that is + // the order `zunionInterDiffGenericCommand` takes: it looks every source + // up (and refuses a wrong type) before it looks at `WEIGHTS`, so + // `ZUNIONSTORE d 1 BOGUS` is `WRONGTYPE` on redis 8.6.1, not + // `syntax error`. Parsing the options first inverted that (moon#959). + // + // Each source is read through the SHARED-borrow view, not `get_sorted_set`: that + // accessor's `get_promoted` core upgrades a listpack source to the B+tree + // form as a side effect of READING it — the moon#928 defect, which the + // read-only set-operation family (`collect_source_sets_readonly`) already + // left behind. A `&Database` borrow cannot reach `SortedSetKind::upgrade` + // at all, and `get_sorted_set_ref_if_alive` classifies every encoding + // (B+tree, listpack, legacy, and a cold-tier hit read through as `Owned`) + // rather than demanding one. Each source is copied into an owned map + // exactly as before, because the destination write below needs `db` + // mutably; a listpack's copy is bounded by `zset-max-listpack-entries`. + let now_ms = db.now_ms(); + let mut source_data: Vec> = Vec::with_capacity(numkeys); + for key in &source_keys { + match db.get_sorted_set_ref_if_alive(key, now_ms) { + Ok(Some(zref)) => match zref.members_map() { + Some(members) => source_data.push(members.clone()), + None => source_data.push(zref.entries_sorted().into_iter().collect()), + }, + Ok(None) => { + source_data.push(HashMap::new()); + } + Err(e) => return e, + } + } + + // Parse WEIGHTS and AGGREGATE + let mut weights: Vec = vec![1.0; numkeys]; + let mut aggregate = AggregateOp::Sum; + let mut i = 2 + numkeys; + + while i < args.len() { + let opt = match extract_bytes(&args[i]) { + Some(b) => b.as_ref(), + None => { + i += 1; + continue; + } + }; + // `WEIGHTS` and `AGGREGATE` are not tokens ZDIFFSTORE knows — Redis + // only matches them `if (op != SET_OP_DIFF)`, so on a diff they fall + // through to the unknown-token arm and are `syntax error` (moon#959). + let takes_weights = op != SetOp::Diff; + if takes_weights && opt.eq_ignore_ascii_case(b"WEIGHTS") { + for w in 0..numkeys { + // Too few weights to cover the key list is `syntax error` on + // Redis, not an arity error (moon#969). + if i + 1 + w >= args.len() { + return err("ERR syntax error"); + } + let wb = match extract_bytes(&args[i + 1 + w]) { + Some(b) => b, + None => return err("ERR syntax error"), + }; + // `"nan"` PARSES in Rust where C's `strtod` + `isnan` check in + // `getDoubleFromObjectOrReply` rejects it (moon#969), so a NaN + // weight sailed through and poisoned every aggregated score. + // Infinities stay legal, as they are on Redis. + let wval: f64 = match std::str::from_utf8(wb) + .ok() + .and_then(|s| s.parse::().ok()) + .filter(|v| !v.is_nan()) + { + Some(v) => v, + None => return err("ERR weight value is not a float"), + }; + weights[w] = wval; + } + i += 1 + numkeys; + } else if takes_weights && opt.eq_ignore_ascii_case(b"AGGREGATE") { + if i + 1 >= args.len() { + return err("ERR syntax error"); + } + let agg_b = match extract_bytes(&args[i + 1]) { + Some(b) => b.as_ref(), + None => return err("ERR syntax error"), + }; + aggregate = if agg_b.eq_ignore_ascii_case(b"SUM") { + AggregateOp::Sum + } else if agg_b.eq_ignore_ascii_case(b"MIN") { + AggregateOp::Min + } else if agg_b.eq_ignore_ascii_case(b"MAX") { + AggregateOp::Max + } else { + return err("ERR syntax error"); + }; + i += 2; + } else { + // moon#967 rewrote every OTHER zset option loop to reject an + // unrecognised token and missed this one, so `ZUNIONSTORE d 1 k + // BOGUS` stepped over `BOGUS` and answered a DIFFERENT, successful + // command. Redis: `ERR syntax error`. + return err("ERR syntax error"); + } + } + + // Compute result + let mut result_map: HashMap = HashMap::new(); + + if op == SetOp::Diff { + // Members of the first source that no later source contains, with the + // first source's scores untouched — the same walk `zdiff_readonly` + // makes. No weight applies: the option loop refused `WEIGHTS`. + if let Some(first) = source_data.first() { + 'outer: for (member, score) in first { + for src in source_data.iter().skip(1) { + if src.contains_key(member) { + continue 'outer; + } + } + result_map.insert(member.clone(), *score); + } + } + } else if op == SetOp::Inter { + // Start with first set's members + if let Some(first) = source_data.first() { + for (member, score) in first { + let weighted = clamp_nan_to_zero(*score * weights[0]); + let mut final_score = weighted; + let mut in_all = true; + + for (idx, src) in source_data.iter().enumerate().skip(1) { + match src.get(member) { + Some(s) => { + let ws = clamp_nan_to_zero(*s * weights[idx]); + final_score = match aggregate { + AggregateOp::Sum => clamp_nan_to_zero(final_score + ws), + AggregateOp::Min => final_score.min(ws), + AggregateOp::Max => final_score.max(ws), + }; + } + None => { + in_all = false; + break; + } + } + } + + if in_all { + result_map.insert(member.clone(), final_score); + } + } + } + } else { + // Union: all members from all sets + for (idx, src) in source_data.iter().enumerate() { + for (member, score) in src { + let weighted = clamp_nan_to_zero(*score * weights[idx]); + result_map + .entry(member.clone()) + .and_modify(|existing| { + *existing = match aggregate { + AggregateOp::Sum => clamp_nan_to_zero(*existing + weighted), + AggregateOp::Min => existing.min(weighted), + AggregateOp::Max => existing.max(weighted), + }; + }) + .or_insert(weighted); + } + } + } + + let result_size = result_map.len() as i64; + + // Remove destination key first, then create new sorted set + db.remove(dest); + + if !result_map.is_empty() { + let (members, scores) = match db.get_or_create_sorted_set(dest) { + Ok(pair) => pair, + Err(e) => return e, + }; + + // `dest` was just removed/recreated above, so every member here is + // new -- charge each unconditionally (O(1) per member, no full + // recompute of the destination sorted set). + let mut mem_charge: usize = 0; + let table_before = zset_table_bytes(members, scores); + for (member, score) in result_map { + mem_charge += zset_member_cost(&member); + zadd_member(members, scores, member, score); + } + let table_after = zset_table_bytes(members, scores); + // `members`/`scores`' borrow of `db` ends above. + db.charge_memory(mem_charge); + db.adjust_memory(table_before, table_after); + } + + Frame::Integer(result_size) +} + +// --------------------------------------------------------------------------- +// ZRANGESTORE dst src min max [BYSCORE | BYLEX] [REV] [LIMIT offset count] +// --------------------------------------------------------------------------- + +/// ZRANGESTORE dst src min max [BYSCORE | BYLEX] [REV] [LIMIT offset count] +/// +/// Stores the result of a ZRANGE into `dst`, replacing it. Returns the cardinality of `dst`. +pub fn zrangestore(db: &mut Database, args: &[Frame]) -> Frame { + if args.len() < 4 { + return err_wrong_args("ZRANGESTORE"); + } + let dst = match extract_bytes(&args[0]) { + Some(k) => k, + None => return err_wrong_args("ZRANGESTORE"), + }; + let src = match extract_bytes(&args[1]) { + Some(k) => k, + None => return err_wrong_args("ZRANGESTORE"), + }; + let min_arg = match extract_bytes(&args[2]) { + Some(b) => b.clone(), + None => return err_wrong_args("ZRANGESTORE"), + }; + let max_arg = match extract_bytes(&args[3]) { + Some(b) => b.clone(), + None => return err_wrong_args("ZRANGESTORE"), + }; + + // Parse optional flags (same as ZRANGE but no WITHSCORES) + let mut by_score = false; + let mut by_lex = false; + let mut rev = false; + let mut limit_offset: Option = None; + let mut limit_count: Option = None; + + let mut i = 4; + while i < args.len() { + let opt = match extract_bytes(&args[i]) { + Some(b) => b.as_ref(), + None => { + i += 1; + continue; + } + }; + if opt.eq_ignore_ascii_case(b"BYSCORE") { + by_score = true; + i += 1; + } else if opt.eq_ignore_ascii_case(b"BYLEX") { + by_lex = true; + i += 1; + } else if opt.eq_ignore_ascii_case(b"REV") { + rev = true; + i += 1; + } else if opt.eq_ignore_ascii_case(b"LIMIT") { + if i + 2 < args.len() { + let off_b = match extract_bytes(&args[i + 1]) { + Some(b) => b, + None => return err_wrong_args("ZRANGESTORE"), + }; + let cnt_b = match extract_bytes(&args[i + 2]) { + Some(b) => b, + None => return err_wrong_args("ZRANGESTORE"), + }; + limit_offset = std::str::from_utf8(off_b).ok().and_then(|s| s.parse().ok()); + limit_count = std::str::from_utf8(cnt_b).ok().and_then(|s| s.parse().ok()); + if limit_offset.is_none() || limit_count.is_none() { + return err("ERR value is not an integer or out of range"); + } + i += 3; + } else { + return err_wrong_args("ZRANGESTORE"); + } + } else { + return err("ERR syntax error"); + } + } + + if by_score && by_lex { + return err("ERR BYSCORE and BYLEX options are not compatible"); + } + if limit_offset.is_some() && !by_score && !by_lex { + return err( + "ERR syntax error, LIMIT is only supported in combination with either BYSCORE or BYLEX", + ); + } + + // Run ZRANGE on src, collecting (member, score) pairs + let entries: Vec<(Bytes, f64)> = match db.get_sorted_set(src) { + Ok(Some((members, scores))) => { + let frame = if by_score { + zrange_by_score( + members, + scores, + &min_arg, + &max_arg, + rev, + true, + limit_offset, + limit_count, + ) + } else if by_lex { + zrange_by_lex( + scores, + &min_arg, + &max_arg, + rev, + true, + members, + limit_offset, + limit_count, + ) + } else { + zrange_by_rank(scores, &min_arg, &max_arg, rev, true) + }; + // Parse the Frame::Array([member, score, member, score, ...]) into Vec<(Bytes, f64)> + match frame { + Frame::Array(arr) => { + let mut result = Vec::with_capacity(arr.len() / 2); + let mut idx = 0; + while idx + 1 < arr.len() { + if let (Frame::BulkString(m), Frame::BulkString(s)) = + (&arr[idx], &arr[idx + 1]) + { + if let Ok(score) = std::str::from_utf8(s).unwrap_or("0").parse::() + { + result.push((m.clone(), score)); + } + } + idx += 2; + } + result + } + Frame::Error(_) => return frame, + _ => Vec::with_capacity(0), + } + } + Ok(None) => Vec::with_capacity(0), + Err(e) => return e, + }; + + let count = entries.len() as i64; + + // Replace dst with the result + db.remove(dst); + + if !entries.is_empty() { + let (dst_members, dst_scores) = match db.get_or_create_sorted_set(dst) { + Ok(pair) => pair, + Err(e) => return e, + }; + // `dst` was just removed/recreated above, so every entry is new. + let mut mem_charge: usize = 0; + let table_before = zset_table_bytes(dst_members, dst_scores); + for (member, score) in entries { + mem_charge += zset_member_cost(&member); + zadd_member(dst_members, dst_scores, member, score); + } + let table_after = zset_table_bytes(dst_members, dst_scores); + // `dst_members`/`dst_scores`' borrow of `db` ends above. + db.charge_memory(mem_charge); + db.adjust_memory(table_before, table_after); + } + + Frame::Integer(count) +} diff --git a/src/command/sorted_set/sorted_set_write.rs b/src/command/sorted_set/sorted_set_write.rs index c0cd5add3..e41bf4ee1 100644 --- a/src/command/sorted_set/sorted_set_write.rs +++ b/src/command/sorted_set/sorted_set_write.rs @@ -1,9 +1,9 @@ use bytes::Bytes; -use std::collections::HashMap; +use ordered_float::OrderedFloat; use crate::protocol::Frame; use crate::storage::Database; -use crate::storage::db::{Shape, zset_member_cost, zset_table_bytes}; +use crate::storage::db::{Shape, SortedSetRef, zset_member_cost, zset_table_bytes}; use crate::storage::listpack::PairUpdate; use crate::storage::zset_score::{ScoreBuf, render_score}; @@ -11,9 +11,9 @@ use crate::command::helpers::{all_args_are_bytes, err, err_wrong_args, extract_b use crate::command::sorted_set::work_budget; use super::{ - AggregateOp, clamp_nan_to_zero, format_score, format_score_bytes, parse_bounded_count, - parse_numkeys, zadd_member, zrange_by_lex, zrange_by_rank, zrange_by_score, zrem_member, - zset_insert_absent, zset_update_existing, + LexBound, ScoreBound, format_score, format_score_bytes, lex_in_range, parse_bounded_count, + parse_lex_bound, parse_score_bound, rank_window, zrem_member, zset_insert_absent, + zset_update_existing, }; // --------------------------------------------------------------------------- @@ -103,7 +103,7 @@ fn resolved_pair<'a>( // through `render_score`); the closures below read it as 0.0 so the member is // still FOUND and updated in place rather than duplicated. -/// ZADD key [NX|XX] [GT|LT] [CH] score member [score member ...] +/// ZADD key [NX|XX] [GT|LT] [CH] [INCR] score member [score member ...] pub fn zadd(db: &mut Database, args: &[Frame]) -> Frame { if args.len() < 3 { return err_wrong_args("ZADD"); @@ -119,6 +119,7 @@ pub fn zadd(db: &mut Database, args: &[Frame]) -> Frame { let mut gt = false; let mut lt = false; let mut ch = false; + let mut incr = false; let mut i = 1; while i < args.len() { @@ -141,6 +142,9 @@ pub fn zadd(db: &mut Database, args: &[Frame]) -> Frame { } else if arg.eq_ignore_ascii_case(b"CH") { ch = true; i += 1; + } else if arg.eq_ignore_ascii_case(b"INCR") { + incr = true; + i += 1; } else { break; } @@ -170,6 +174,26 @@ pub fn zadd(db: &mut Database, args: &[Frame]) -> Frame { return err("ERR syntax error"); } + // `INCR` (moon#959): ZINCRBY's arithmetic under ZADD's flags, replying the + // new score as a bulk string, or nil when a flag refused the write. Redis + // checks the pair count AFTER the parity and flag-pairing rules above, so + // `ZADD k INCR 1` is `syntax error` and `ZADD k INCR NX XX 1 a 2 b` is the + // NX/XX error — both verified on redis 8.6.1. The increment is parsed by + // the ONE parser every `score member` pair goes through, so a NaN or a + // non-float is refused with `ZADD`'s own message before the keyspace is + // touched. + if incr { + if remaining.len() != 2 { + return err("ERR INCR option supports a single increment-element pair"); + } + let (increment, member) = match parse_zadd_pair(&remaining[0], &remaining[1]) { + Ok(pair) => pair, + Err(e) => return e, + }; + // `CH` has no effect on the INCR reply, as on Redis. + return zincr_member(db, key, increment, member, IncrFlags { nx, xx, gt, lt }); + } + // moon#814: validate EVERY pair BEFORE touching the keyspace. // // The mutation loop below runs inside the `table_before … charge_memory()` @@ -628,6 +652,229 @@ pub fn zrem(db: &mut Database, args: &[Frame]) -> Frame { Frame::Integer(removed) } +// --------------------------------------------------------------------------- +// ZREMRANGEBYRANK / ZREMRANGEBYSCORE / ZREMRANGEBYLEX (moon#959) +// --------------------------------------------------------------------------- + +/// The window a `ZREMRANGEBY*` command deletes, parsed BEFORE the keyspace is +/// touched so a bad bound never fabricates or reclaims a key — Redis parses +/// the range first and only then looks the key up, so `ZREMRANGEBYSCORE +/// nokey a 1` is `min or max is not a float` and not `0`. +enum RemRange { + Rank(i64, i64), + Score(ScoreBound, ScoreBound), + Lex(LexBound, LexBound), +} + +impl RemRange { + /// The members of a score-sorted decode that fall inside the window. + /// Borrowed from `entries`, which is the caller's own copy, so the + /// listpack they came from can be mutated while these are consumed. + fn select<'a>(&self, entries: &'a [(Bytes, f64)]) -> Vec<&'a Bytes> { + match self { + RemRange::Rank(start, stop) => match rank_window(*start, *stop, entries.len()) { + Some((lo, hi)) => entries[lo..=hi].iter().map(|(m, _)| m).collect(), + None => Vec::new(), + }, + RemRange::Score(min, max) => entries + .iter() + .filter(|(_, s)| min.includes(*s) && max.includes_upper(*s)) + .map(|(m, _)| m) + .collect(), + RemRange::Lex(min, max) => entries + .iter() + .filter(|(m, _)| lex_in_range(m, min, max)) + .map(|(m, _)| m) + .collect(), + } + } +} + +/// ZREMRANGEBYRANK key start stop +pub fn zremrangebyrank(db: &mut Database, args: &[Frame]) -> Frame { + let (key, min_b, max_b) = match zremrange_args(args, "ZREMRANGEBYRANK") { + Ok(v) => v, + Err(e) => return e, + }; + // A rank index is read with a NULL message on Redis, so the generic + // integer error is the right class here (moon#969 documents the same for + // ZRANGE's indices). + let parse = |b: &[u8]| -> Result { + std::str::from_utf8(b) + .ok() + .and_then(|s| s.parse().ok()) + .ok_or_else(|| err("ERR value is not an integer or out of range")) + }; + let (start, stop) = match (parse(min_b), parse(max_b)) { + (Ok(a), Ok(b)) => (a, b), + (Err(e), _) | (_, Err(e)) => return e, + }; + zremrange_impl(db, key, RemRange::Rank(start, stop)) +} + +/// ZREMRANGEBYSCORE key min max +pub fn zremrangebyscore(db: &mut Database, args: &[Frame]) -> Frame { + let (key, min_b, max_b) = match zremrange_args(args, "ZREMRANGEBYSCORE") { + Ok(v) => v, + Err(e) => return e, + }; + let (min, max) = match (parse_score_bound(min_b), parse_score_bound(max_b)) { + (Ok(a), Ok(b)) => (a, b), + (Err(e), _) | (_, Err(e)) => return e, + }; + zremrange_impl(db, key, RemRange::Score(min, max)) +} + +/// ZREMRANGEBYLEX key min max +pub fn zremrangebylex(db: &mut Database, args: &[Frame]) -> Frame { + let (key, min_b, max_b) = match zremrange_args(args, "ZREMRANGEBYLEX") { + Ok(v) => v, + Err(e) => return e, + }; + let (min, max) = match (parse_lex_bound(min_b), parse_lex_bound(max_b)) { + (Ok(a), Ok(b)) => (a, b), + (Err(e), _) | (_, Err(e)) => return e, + }; + zremrange_impl(db, key, RemRange::Lex(min, max)) +} + +/// The `key min max` shape all three share: exactly three arguments (their +/// registered arity is 4), every one a bulk string. +fn zremrange_args<'a>( + args: &'a [Frame], + cmd: &'static str, +) -> Result<(&'a Bytes, &'a Bytes, &'a Bytes), Frame> { + if args.len() != 3 { + return Err(err_wrong_args(cmd)); + } + match ( + extract_bytes(&args[0]), + extract_bytes(&args[1]), + extract_bytes(&args[2]), + ) { + (Some(k), Some(min), Some(max)) => Ok((k, min, max)), + _ => Err(err_wrong_args(cmd)), + } +} + +/// Delete every member inside `range` and reply how many went. +/// +/// The same two-arm shape as `zrem` (moon#897): a listpack is trimmed in +/// place and never converted — a removal cannot cross a threshold upward — +/// and the B+tree arm credits each member's cost and the table shrink exactly +/// as `zrem` does. A key that drains to empty is removed on both arms, which +/// is also what reclaims the empty container `get_or_create_zset_listpack` +/// fabricates for a missing key, so `ZREMRANGEBYRANK nokey 0 1` answers `0` +/// and leaves no key behind. +/// +/// The victims are materialised before the first removal on both arms: a +/// listpack keeps insertion order, so the window is decided on a score-sorted +/// decode (bounded by `zset-max-listpack-entries`), and a B+tree cannot be +/// mutated while its iterator is live. The B+tree list holds `Bytes` handles, +/// which are reference-count bumps rather than copies. +fn zremrange_impl(db: &mut Database, key: &[u8], range: RemRange) -> Frame { + match db.get_or_create_zset_listpack(key) { + Ok(Some(lp)) => { + // Listpack `estimate_memory()` is O(1) (capacity-based). + let before = lp.estimate_memory(); + let entries = SortedSetRef::Listpack(&*lp).entries_sorted(); + let mut removed = 0i64; + for member in range.select(&entries) { + // `remove_pair` matches the FIELD half only — the member, + // never the score — and drains both entries in one scan. + if lp.remove_pair(member) { + removed += 1; + } + } + let after = lp.estimate_memory(); + let is_empty = lp.is_empty(); + // `lp`'s borrow of `db` ends here. + db.adjust_memory(before, after); + if is_empty { + db.remove(key); + } + return Frame::Integer(removed); + } + // Already the full B+tree form (or a cold-promoted value, which never + // decodes compact): fall through. + Ok(None) => {} + Err(e) => return e, // WRONGTYPE + } + + let (members, scores) = match db.get_or_create_sorted_set(key) { + Ok(pair) => pair, + Err(e) => return e, + }; + + let victims: Vec = match &range { + RemRange::Rank(start, stop) => match rank_window(*start, *stop, scores.len()) { + Some((lo, hi)) => scores + .range_by_rank(lo, hi) + .into_iter() + .map(|(_, m)| m.clone()) + .collect(), + None => Vec::new(), + }, + RemRange::Score(min, max) => { + // `BPTree::range` wants `lo <= hi`; a reversed pair is an empty + // window on Redis (`ZREMRANGEBYSCORE k 3 1` removes nothing), and + // the bound filters keep an exclusive or infinite edge exact. + let lo = OrderedFloat(min.value()); + let hi = OrderedFloat(max.value()); + if lo > hi { + Vec::new() + } else { + scores + .range(lo, hi) + .filter(|(s, _)| min.includes(s.0) && max.includes_upper(s.0)) + .map(|(_, m)| m.clone()) + .collect() + } + } + RemRange::Lex(min, max) => scores + .iter() + .filter(|(_, m)| lex_in_range(m, min, max)) + .map(|(_, m)| m.clone()) + .collect(), + }; + + let mut removed = 0i64; + let mut credit: usize = 0; + let table_before = zset_table_bytes(members, scores); + for member in &victims { + if zrem_member(members, scores, member) { + removed += 1; + credit += zset_member_cost(member); + } + } + let is_empty = members.is_empty(); + let table_after = zset_table_bytes(members, scores); + // `members`/`scores`' borrow of `db` ends above. + db.credit_memory(credit); + // Unconditional, as in `zrem`: `db.remove` credits `entry_overhead` + // recomputed from the CURRENT value, and a shrunken table's capacity + // must be credited here or it is stranded. + db.adjust_memory(table_before, table_after); + if is_empty { + db.remove(key); + } + Frame::Integer(removed) +} + +// --------------------------------------------------------------------------- +// ZINCRBY, and the arithmetic core it shares with `ZADD ... INCR` +// --------------------------------------------------------------------------- + +/// The `ZADD` flags that bear on an increment (moon#959). All false for a +/// plain `ZINCRBY`. +#[derive(Debug, Clone, Copy, Default)] +struct IncrFlags { + nx: bool, + xx: bool, + gt: bool, + lt: bool, +} + /// ZINCRBY key increment member pub fn zincrby(db: &mut Database, args: &[Frame]) -> Frame { if args.len() != 3 { @@ -642,7 +889,7 @@ pub fn zincrby(db: &mut Database, args: &[Frame]) -> Frame { None => return err_wrong_args("ZINCRBY"), }; let member = match extract_bytes(&args[2]) { - Some(b) => b.clone(), + Some(b) => b, None => return err_wrong_args("ZINCRBY"), }; @@ -658,6 +905,27 @@ pub fn zincrby(db: &mut Database, args: &[Frame]) -> Frame { return err("ERR value is not a valid float"); } + zincr_member(db, key, increment, member, IncrFlags::default()) +} + +/// Add `increment` to `member`'s score, creating the member at `increment` +/// when it is absent, and reply the new score — or nil when a flag refused +/// the write. +/// +/// The decision order is Redis's `zsetAdd` with `ZADD_IN_INCR`, verified on +/// redis 8.6.1: for a PRESENT member, `NX` refuses before the sum is even +/// formed; then a NaN sum is `ERR resulting score is not a number (NaN)` with +/// nothing written; then `GT`/`LT` refuse a sum that does not move the score +/// the right way (`GT` with a zero increment is a refusal). For an ABSENT +/// member only `XX` refuses; `GT`/`LT` never block a first insert. A refusal +/// on a key this call had to fabricate leaves no key behind. +fn zincr_member( + db: &mut Database, + key: &[u8], + increment: f64, + member: &Bytes, + flags: IncrFlags, +) -> Frame { // Listpack path (moon#897). ZINCRBY is the leaderboard primitive, and // before this it took the eager `get_or_create_sorted_set`: one ZINCRBY // flattened a three-member zset to `skiplist` permanently (nothing @@ -677,6 +945,12 @@ pub fn zincrby(db: &mut Database, args: &[Frame]) -> Frame { Ok(Some(lp)) => { // Listpack `estimate_memory()` is O(1) (capacity-based). let before = lp.estimate_memory(); + // Why the closure declined, when it did. `Unchanged` alone + // cannot say: it is a NaN sum for a plain ZINCRBY and a flag + // refusal under `ZADD ... INCR`, and the two reply + // differently. + let mut reached_nan = false; + let mut refused = false; // ONE scan (moon#942): the walk that finds the member carries // the byte offsets its score is rewritten at, so there is no // second walk back to an ordinal. @@ -685,23 +959,25 @@ pub fn zincrby(db: &mut Database, args: &[Frame]) -> Frame { // its RENDERED text, and `render_score(NaN)` writes `NaN`, // which `parse_score` refuses — the score would read back as // 0.0 and the member would silently change value. `increment` - // is already proven non-NaN above, so this is reachable only - // as `±inf + ∓inf`, which in turn means the member already - // exists (a fresh member starts at 0.0) — so declining here - // never leaves a key created-and-abandoned. - // - // Declining hands the case to the B+tree arm below, which is - // byte-for-byte what EVERY ZINCRBY did before this branch - // existed. moon's reply there (`NaN`) diverges from redis - // 8.6.1, which answers - // `ERR resulting score is not a number (NaN)` and leaves the - // score untouched — a real, PRE-EXISTING divergence that this - // change deliberately does not alter, and that a NaN must - // never reach a listpack in the meantime. - let outcome = lp.update_pair_value(&member, |current| { + // is already proven non-NaN by every caller, so this is + // reachable only as `±inf + ∓inf`, which in turn means the + // member already exists (a fresh member starts at 0.0) — so + // declining here never leaves a key created-and-abandoned. + let outcome = lp.update_pair_value(member, |current| { + if flags.nx { + // NX: the member is present, which is all it needs. + refused = true; + return None; + } work_budget::note_stored_score_parse(); - let new_score = current.as_score().unwrap_or(0.0) + increment; + let old = current.as_score().unwrap_or(0.0); + let new_score = old + increment; if new_score.is_nan() { + reached_nan = true; + return None; + } + if (flags.gt && new_score <= old) || (flags.lt && new_score >= old) { + refused = true; return None; } // One stack buffer; `render_score` is byte-identical to @@ -718,17 +994,38 @@ pub fn zincrby(db: &mut Database, args: &[Frame]) -> Frame { // reply below does not render the score a second time. PairUpdate::Replaced(rendered) => Some(rendered), PairUpdate::Absent => { - // A fresh member starts at 0.0 and `increment` is - // already proven non-NaN, so this rendering can never - // be the NaN the arm above guards against. - let mut rendered = ScoreBuf::new(); - render_score(increment, &mut rendered); - lp.push_back(&member); - lp.push_back(&rendered); - Some(rendered) + if flags.xx { + // XX: never create. The container may be one + // this call fabricated; the empty check below + // reclaims it. + refused = true; + None + } else { + // A fresh member starts at 0.0 and `increment` is + // already proven non-NaN, so this rendering can + // never be the NaN the arm above guards against. + let mut rendered = ScoreBuf::new(); + render_score(increment, &mut rendered); + lp.push_back(member); + lp.push_back(&rendered); + Some(rendered) + } } - // NaN. The listpack was not touched, and this is the - // answer (moon#960): redis 8.6.1 replies + // NaN, or a flag refusal: the listpack was not touched. + PairUpdate::Unchanged => None, + }; + + let after = lp.estimate_memory(); + // The upgrade check, from the same authority as the gate: + // it converts `lp.len()` (member AND score entries) to + // members itself — the moon#896 unit. + let should_upgrade = + stored.is_some() && !limits.listpack_fits(Shape::SortedSet, lp); + let is_empty = lp.is_empty(); + // `lp`'s borrow of `db` ends here. + db.adjust_memory(before, after); + if reached_nan { + // This is the answer (moon#960): redis 8.6.1 replies // `ERR resulting score is not a number (NaN)` and leaves // the score alone. Returning here rather than falling // through also keeps the encoding intact — the B+tree arm @@ -736,36 +1033,38 @@ pub fn zincrby(db: &mut Database, args: &[Frame]) -> Frame { // so falling through would flatten the listpack // permanently (moon#832: nothing demotes) as a side // effect of a command that errors and stores nothing. - PairUpdate::Unchanged => { - return err("ERR resulting score is not a number (NaN)"); - } - }; - + return err("ERR resulting score is not a number (NaN)"); + } + if is_empty { + // Only reachable as an `XX` refusal on a fabricated + // container — the same rule `zadd` applies. + db.remove(key); + } + if refused { + return Frame::Null; + } + if should_upgrade { + // Self-accounting: the accessor bills the one-time + // listpack -> B+tree swing itself (moon#788/#810). + db.upgrade_zset_listpack_to_bptree(key); + } + // Reply with the bytes we STORED, not a second rendering: + // one copy out of the stack buffer instead of the + // `format_score` -> `String` allocation the B+tree arm + // below still pays (`src/command/` is a no-`String` + // path). `render_score` is pinned byte-identical to + // `format_score_bytes` by + // `listpack_score_rendering_matches_zscore_rendering`, so + // this is the same text either way — and it is now the + // same text a later ZSCORE reads out of the listpack, by + // construction rather than by two formatters agreeing. if let Some(rendered) = stored { - let after = lp.estimate_memory(); - // The upgrade check, from the same authority as the gate: - // it converts `lp.len()` (member AND score entries) to - // members itself — the moon#896 unit. - let should_upgrade = !limits.listpack_fits(Shape::SortedSet, lp); - // `lp`'s borrow of `db` ends here. - db.adjust_memory(before, after); - if should_upgrade { - // Self-accounting: the accessor bills the one-time - // listpack -> B+tree swing itself (moon#788/#810). - db.upgrade_zset_listpack_to_bptree(key); - } - // Reply with the bytes we STORED, not a second rendering: - // one copy out of the stack buffer instead of the - // `format_score` -> `String` allocation the B+tree arm - // below still pays (`src/command/` is a no-`String` - // path). `render_score` is pinned byte-identical to - // `format_score_bytes` by - // `listpack_score_rendering_matches_zscore_rendering`, so - // this is the same text either way — and it is now the - // same text a later ZSCORE reads out of the listpack, by - // construction rather than by two formatters agreeing. return Frame::BulkString(Bytes::copy_from_slice(&rendered)); } + // `stored` is `None` exactly when `reached_nan || refused`, + // both returned above. Kept as a real match rather than an + // `unwrap`, as the mutation loops in `zadd` are. + return Frame::Null; } // Already the full B+tree form (or a cold-promoted value, which // never decodes compact): fall through. @@ -779,7 +1078,7 @@ pub fn zincrby(db: &mut Database, args: &[Frame]) -> Frame { Err(e) => return e, }; - let member_cost = zset_member_cost(&member); + let member_cost = zset_member_cost(member); let table_before = zset_table_bytes(members, scores); // ONE hash lookup (moon#942): the same lookup that reads the current score // writes `current + increment` back through the slot it found. It used to @@ -791,12 +1090,21 @@ pub fn zincrby(db: &mut Database, args: &[Frame]) -> Frame { // from the closure is `zset_update_existing`'s "leave it alone", so the // old score survives and neither map is touched. let mut reached_nan = false; - let is_new = zset_update_existing(members, scores, &member, |current| { + let mut refused = false; + let is_new = zset_update_existing(members, scores, member, |current| { + if flags.nx { + refused = true; + return None; + } let candidate = current + increment; if candidate.is_nan() { reached_nan = true; return None; } + if (flags.gt && candidate <= current) || (flags.lt && candidate >= current) { + refused = true; + return None; + } new_score = candidate; Some(candidate) }) @@ -804,19 +1112,28 @@ pub fn zincrby(db: &mut Database, args: &[Frame]) -> Frame { if reached_nan { return err("ERR resulting score is not a number (NaN)"); } - if is_new { + let inserted = is_new && !flags.xx; + if inserted { // A member that was not there starts at 0.0, so its new score is the // increment itself — already in `new_score`. - zset_insert_absent(members, scores, member, new_score); + zset_insert_absent(members, scores, member.clone(), new_score); } + let is_empty = members.is_empty(); let table_after = zset_table_bytes(members, scores); // `members`/`scores`' borrow of `db` ends above. - if is_new { + if inserted { db.charge_memory(member_cost); } db.adjust_memory(table_before, table_after); + if is_empty { + // `XX` refused the only member a fabricated container would have had. + db.remove(key); + } + if refused || (is_new && flags.xx) { + return Frame::Null; + } - Frame::BulkString(Bytes::from(format_score(new_score))) + Frame::BulkString(format_score_bytes(new_score)) } /// ZPOPMIN key [count] @@ -961,381 +1278,6 @@ pub fn zpopmax(db: &mut Database, args: &[Frame]) -> Frame { Frame::Array(result.into()) } -/// ZUNIONSTORE destination numkeys key [key ...] [WEIGHTS weight ...] [AGGREGATE SUM|MIN|MAX] -pub fn zunionstore(db: &mut Database, args: &[Frame]) -> Frame { - zstore_impl(db, args, false) -} - -/// ZINTERSTORE destination numkeys key [key ...] [WEIGHTS weight ...] [AGGREGATE SUM|MIN|MAX] -pub fn zinterstore(db: &mut Database, args: &[Frame]) -> Frame { - zstore_impl(db, args, true) -} - -fn zstore_impl(db: &mut Database, args: &[Frame], intersect: bool) -> Frame { - let cmd_name = if intersect { - "ZINTERSTORE" - } else { - "ZUNIONSTORE" - }; - if args.len() < 3 { - return err_wrong_args(cmd_name); - } - let dest = match extract_bytes(&args[0]) { - Some(k) => k, - None => return err_wrong_args(cmd_name), - }; - let numkeys_bytes = match extract_bytes(&args[1]) { - Some(b) => b, - None => return err_wrong_args(cmd_name), - }; - let numkeys = match parse_numkeys(numkeys_bytes, cmd_name) { - Ok(n) => n, - Err(e) => return e, - }; - - // A `numkeys` that overruns the key list is `syntax error`, not an arity - // error (moon#969) — Redis's arity check already passed above, and - // `zunionInterDiffGenericCommand` answers `shared.syntaxerr` here. - if args.len() < 2 + numkeys { - return err("ERR syntax error"); - } - - // Collect source keys - let source_keys: Vec = (0..numkeys) - .map(|j| { - extract_bytes(&args[2 + j]) - .cloned() - .unwrap_or_else(|| Bytes::new()) - }) - .collect(); - - // Parse WEIGHTS and AGGREGATE - let mut weights: Vec = vec![1.0; numkeys]; - let mut aggregate = AggregateOp::Sum; - let mut i = 2 + numkeys; - - while i < args.len() { - let opt = match extract_bytes(&args[i]) { - Some(b) => b.as_ref(), - None => { - i += 1; - continue; - } - }; - if opt.eq_ignore_ascii_case(b"WEIGHTS") { - for w in 0..numkeys { - // Too few weights to cover the key list is `syntax error` on - // Redis, not an arity error (moon#969). - if i + 1 + w >= args.len() { - return err("ERR syntax error"); - } - let wb = match extract_bytes(&args[i + 1 + w]) { - Some(b) => b, - None => return err("ERR syntax error"), - }; - // `"nan"` PARSES in Rust where C's `strtod` + `isnan` check in - // `getDoubleFromObjectOrReply` rejects it (moon#969), so a NaN - // weight sailed through and poisoned every aggregated score. - // Infinities stay legal, as they are on Redis. - let wval: f64 = match std::str::from_utf8(wb) - .ok() - .and_then(|s| s.parse::().ok()) - .filter(|v| !v.is_nan()) - { - Some(v) => v, - None => return err("ERR weight value is not a float"), - }; - weights[w] = wval; - } - i += 1 + numkeys; - } else if opt.eq_ignore_ascii_case(b"AGGREGATE") { - if i + 1 >= args.len() { - return err("ERR syntax error"); - } - let agg_b = match extract_bytes(&args[i + 1]) { - Some(b) => b.as_ref(), - None => return err("ERR syntax error"), - }; - aggregate = if agg_b.eq_ignore_ascii_case(b"SUM") { - AggregateOp::Sum - } else if agg_b.eq_ignore_ascii_case(b"MIN") { - AggregateOp::Min - } else if agg_b.eq_ignore_ascii_case(b"MAX") { - AggregateOp::Max - } else { - return err("ERR syntax error"); - }; - i += 2; - } else { - // moon#967 rewrote every OTHER zset option loop to reject an - // unrecognised token and missed this one, so `ZUNIONSTORE d 1 k - // BOGUS` stepped over `BOGUS` and answered a DIFFERENT, successful - // command. Redis: `ERR syntax error`. - return err("ERR syntax error"); - } - } - - // Read all source sets into a temporary structure - let mut source_data: Vec> = Vec::with_capacity(numkeys); - for key in &source_keys { - match db.get_sorted_set(key) { - Ok(Some((members, _))) => { - source_data.push(members.clone()); - } - Ok(None) => { - source_data.push(HashMap::new()); - } - Err(e) => return e, - } - } - - // Compute result - let mut result_map: HashMap = HashMap::new(); - - if intersect { - // Start with first set's members - if let Some(first) = source_data.first() { - for (member, score) in first { - let weighted = clamp_nan_to_zero(*score * weights[0]); - let mut final_score = weighted; - let mut in_all = true; - - for (idx, src) in source_data.iter().enumerate().skip(1) { - match src.get(member) { - Some(s) => { - let ws = clamp_nan_to_zero(*s * weights[idx]); - final_score = match aggregate { - AggregateOp::Sum => clamp_nan_to_zero(final_score + ws), - AggregateOp::Min => final_score.min(ws), - AggregateOp::Max => final_score.max(ws), - }; - } - None => { - in_all = false; - break; - } - } - } - - if in_all { - result_map.insert(member.clone(), final_score); - } - } - } - } else { - // Union: all members from all sets - for (idx, src) in source_data.iter().enumerate() { - for (member, score) in src { - let weighted = clamp_nan_to_zero(*score * weights[idx]); - result_map - .entry(member.clone()) - .and_modify(|existing| { - *existing = match aggregate { - AggregateOp::Sum => clamp_nan_to_zero(*existing + weighted), - AggregateOp::Min => existing.min(weighted), - AggregateOp::Max => existing.max(weighted), - }; - }) - .or_insert(weighted); - } - } - } - - let result_size = result_map.len() as i64; - - // Remove destination key first, then create new sorted set - db.remove(dest); - - if !result_map.is_empty() { - let (members, scores) = match db.get_or_create_sorted_set(dest) { - Ok(pair) => pair, - Err(e) => return e, - }; - - // `dest` was just removed/recreated above, so every member here is - // new -- charge each unconditionally (O(1) per member, no full - // recompute of the destination sorted set). - let mut mem_charge: usize = 0; - let table_before = zset_table_bytes(members, scores); - for (member, score) in result_map { - mem_charge += zset_member_cost(&member); - zadd_member(members, scores, member, score); - } - let table_after = zset_table_bytes(members, scores); - // `members`/`scores`' borrow of `db` ends above. - db.charge_memory(mem_charge); - db.adjust_memory(table_before, table_after); - } - - Frame::Integer(result_size) -} - -// --------------------------------------------------------------------------- -// ZRANGESTORE dst src min max [BYSCORE | BYLEX] [REV] [LIMIT offset count] -// --------------------------------------------------------------------------- - -/// ZRANGESTORE dst src min max [BYSCORE | BYLEX] [REV] [LIMIT offset count] -/// -/// Stores the result of a ZRANGE into `dst`, replacing it. Returns the cardinality of `dst`. -pub fn zrangestore(db: &mut Database, args: &[Frame]) -> Frame { - if args.len() < 4 { - return err_wrong_args("ZRANGESTORE"); - } - let dst = match extract_bytes(&args[0]) { - Some(k) => k, - None => return err_wrong_args("ZRANGESTORE"), - }; - let src = match extract_bytes(&args[1]) { - Some(k) => k, - None => return err_wrong_args("ZRANGESTORE"), - }; - let min_arg = match extract_bytes(&args[2]) { - Some(b) => b.clone(), - None => return err_wrong_args("ZRANGESTORE"), - }; - let max_arg = match extract_bytes(&args[3]) { - Some(b) => b.clone(), - None => return err_wrong_args("ZRANGESTORE"), - }; - - // Parse optional flags (same as ZRANGE but no WITHSCORES) - let mut by_score = false; - let mut by_lex = false; - let mut rev = false; - let mut limit_offset: Option = None; - let mut limit_count: Option = None; - - let mut i = 4; - while i < args.len() { - let opt = match extract_bytes(&args[i]) { - Some(b) => b.as_ref(), - None => { - i += 1; - continue; - } - }; - if opt.eq_ignore_ascii_case(b"BYSCORE") { - by_score = true; - i += 1; - } else if opt.eq_ignore_ascii_case(b"BYLEX") { - by_lex = true; - i += 1; - } else if opt.eq_ignore_ascii_case(b"REV") { - rev = true; - i += 1; - } else if opt.eq_ignore_ascii_case(b"LIMIT") { - if i + 2 < args.len() { - let off_b = match extract_bytes(&args[i + 1]) { - Some(b) => b, - None => return err_wrong_args("ZRANGESTORE"), - }; - let cnt_b = match extract_bytes(&args[i + 2]) { - Some(b) => b, - None => return err_wrong_args("ZRANGESTORE"), - }; - limit_offset = std::str::from_utf8(off_b).ok().and_then(|s| s.parse().ok()); - limit_count = std::str::from_utf8(cnt_b).ok().and_then(|s| s.parse().ok()); - if limit_offset.is_none() || limit_count.is_none() { - return err("ERR value is not an integer or out of range"); - } - i += 3; - } else { - return err_wrong_args("ZRANGESTORE"); - } - } else { - return err("ERR syntax error"); - } - } - - if by_score && by_lex { - return err("ERR BYSCORE and BYLEX options are not compatible"); - } - if limit_offset.is_some() && !by_score && !by_lex { - return err( - "ERR syntax error, LIMIT is only supported in combination with either BYSCORE or BYLEX", - ); - } - - // Run ZRANGE on src, collecting (member, score) pairs - let entries: Vec<(Bytes, f64)> = match db.get_sorted_set(src) { - Ok(Some((members, scores))) => { - let frame = if by_score { - zrange_by_score( - members, - scores, - &min_arg, - &max_arg, - rev, - true, - limit_offset, - limit_count, - ) - } else if by_lex { - zrange_by_lex( - scores, - &min_arg, - &max_arg, - rev, - true, - members, - limit_offset, - limit_count, - ) - } else { - zrange_by_rank(scores, &min_arg, &max_arg, rev, true) - }; - // Parse the Frame::Array([member, score, member, score, ...]) into Vec<(Bytes, f64)> - match frame { - Frame::Array(arr) => { - let mut result = Vec::with_capacity(arr.len() / 2); - let mut idx = 0; - while idx + 1 < arr.len() { - if let (Frame::BulkString(m), Frame::BulkString(s)) = - (&arr[idx], &arr[idx + 1]) - { - if let Ok(score) = std::str::from_utf8(s).unwrap_or("0").parse::() - { - result.push((m.clone(), score)); - } - } - idx += 2; - } - result - } - Frame::Error(_) => return frame, - _ => Vec::with_capacity(0), - } - } - Ok(None) => Vec::with_capacity(0), - Err(e) => return e, - }; - - let count = entries.len() as i64; - - // Replace dst with the result - db.remove(dst); - - if !entries.is_empty() { - let (dst_members, dst_scores) = match db.get_or_create_sorted_set(dst) { - Ok(pair) => pair, - Err(e) => return e, - }; - // `dst` was just removed/recreated above, so every entry is new. - let mut mem_charge: usize = 0; - let table_before = zset_table_bytes(dst_members, dst_scores); - for (member, score) in entries { - mem_charge += zset_member_cost(&member); - zadd_member(dst_members, dst_scores, member, score); - } - let table_after = zset_table_bytes(dst_members, dst_scores); - // `dst_members`/`dst_scores`' borrow of `db` ends above. - db.charge_memory(mem_charge); - db.adjust_memory(table_before, table_after); - } - - Frame::Integer(count) -} - // --------------------------------------------------------------------------- // ZMPOP numkeys key [key ...] MIN|MAX [COUNT n] // --------------------------------------------------------------------------- diff --git a/src/server/conn/shared.rs b/src/server/conn/shared.rs index 63864db94..cb1114c6d 100644 --- a/src/server/conn/shared.rs +++ b/src/server/conn/shared.rs @@ -2141,14 +2141,6 @@ pub(crate) const CROSS_SHARD_WRITE_ERROR: &[u8] = /// entry points (`blocking::immediate_scan`, `blocking::wakeup`) that this /// pre-routing guard cannot see. Two overlapping guards for one family would /// be worse than one complete one. -/// * `ZDIFFSTORE` — not implemented in moon (unknown command), so there is -/// no write to misplace, and claiming `CROSSSLOT` would send a client -/// chasing hash tags for a command that will never work. -/// `tests/two_key_write_cross_shard.rs::t2k4` fails the moment it starts -/// working, which is when it must be added here. `GEORADIUS`/ -/// `GEORADIUSBYMEMBER` used to sit in this same bucket; moon#645 -/// implemented their `STORE`/`STOREDIST` clause, so they moved INTO the -/// family below in the same change that made them able to write. /// * `TOUCH` — the one member of the moon#962 family that is genuinely /// per-key decomposable. It is in `is_multi_key_command` and /// [`splittable_read_kind`], so it FANS OUT and sums, exactly like `EXISTS`. @@ -2195,6 +2187,24 @@ pub(crate) const CROSS_SHARD_WRITE_ERROR: &[u8] = /// this list once it merges properly; each such change removes an error and /// cannot regress correctness, which is the direction that is safe to defer. /// +/// # moon#959 — `ZDIFFSTORE` is IN the family, and used not to be +/// +/// This block used to carry a `ZDIFFSTORE` bullet in the EXCLUDED list above, +/// reading "not implemented in moon (unknown command), so there is no write to +/// misplace". moon#959 implemented it, so that sentence is now false and the +/// bullet is gone: `ZDIFFSTORE dst numkeys src ...` routes on `dst` and reads +/// every source, the identical shape to `ZUNIONSTORE`/`ZINTERSTORE`. It is +/// matched in the `(10, b'z')` arm below, which it SHARES with `ZINTERCARD` — +/// same length, same first byte, so an arm that names only one of them +/// silently drops the other. +/// +/// `GEORADIUS`/`GEORADIUSBYMEMBER` made the same trip when moon#645 gave them +/// a `STORE`/`STOREDIST` clause. The tripwire that forces the migration is +/// `tests/two_key_write_cross_shard.rs::t2k4`, and the measured cost of +/// skipping it is in `t2k1`: with the `ZDIFFSTORE` spelling removed from the +/// arm below and everything else in place, 12 of 180 placements at +/// `--shards 4` ack `ZDIFFSTORE` while the destination lands nowhere. +/// /// Matched on `(len, first byte)` first so a single-key command falls through /// after one integer compare and never reaches the key walk. fn touches_a_key_it_did_not_route_on(cmd: &[u8]) -> bool { @@ -2244,7 +2254,15 @@ fn touches_a_key_it_did_not_route_on(cmd: &[u8]) -> bool { (5, b'z') => cmd.eq_ignore_ascii_case(b"ZMPOP") || cmd.eq_ignore_ascii_case(b"ZDIFF"), (6, b's') => cmd.eq_ignore_ascii_case(b"SINTER") || cmd.eq_ignore_ascii_case(b"SUNION"), (6, b'z') => cmd.eq_ignore_ascii_case(b"ZINTER") || cmd.eq_ignore_ascii_case(b"ZUNION"), - (10, b'z') => cmd.eq_ignore_ascii_case(b"ZINTERCARD"), + // One arm, two unrelated additions: `ZINTERCARD` is a moon#962 + // multi-key READ, `ZDIFFSTORE` a moon#959 two-key WRITE routed on its + // destination. They collide on `(10, b'z')`, so naming only one of + // them here silently drops the other from the guard — for + // `ZDIFFSTORE` that is the moon#592 misdirected write, measured in + // `t2k1`. Keep both spellings. + (10, b'z') => { + cmd.eq_ignore_ascii_case(b"ZINTERCARD") || cmd.eq_ignore_ascii_case(b"ZDIFFSTORE") + } (14, b'g') => cmd.eq_ignore_ascii_case(b"GEOSEARCHSTORE"), // `GEORADIUS src ... STORE|STOREDIST dst` (moon#645). Without the // clause the walker reports one key and this check is a no-op, so no @@ -5078,6 +5096,12 @@ mod cross_shard_write_tests { ("ZRANGESTORE", &["{d}", "{s}", "0", "-1"]), ("ZUNIONSTORE", &["{d}", "1", "{s}"]), ("ZINTERSTORE", &["{d}", "1", "{s}"]), + // moon#959 implemented ZDIFFSTORE. Until it did, the test below + // asserted the OPPOSITE — that the guard must not claim it — because + // an unimplemented command has no write to misplace. It shares the + // `(10, b'z')` arm with `ZINTERCARD`, so this row is what fails if a + // future edit narrows that arm back to one spelling. + ("ZDIFFSTORE", &["{d}", "1", "{s}"]), ("PFMERGE", &["{d}", "{s}"]), ( "GEOSEARCHSTORE", @@ -5234,15 +5258,11 @@ mod cross_shard_write_tests { cross_shard_multikey_rejection(b"TOUCH", &two, N).is_none(), "TOUCH is per-key decomposable and must fan out, never be refused" ); - // ZDIFFSTORE is still unimplemented, and shares a `(10, 'z')` arm with - // ZINTERCARD. Claiming CROSSSLOT for it would send a client chasing - // hash tags for a command that will never work — the `t2k4` tripwire - // in `tests/two_key_write_cross_shard.rs` owns the migration. - assert!( - cross_shard_multikey_rejection(b"ZDIFFSTORE", &[bulk(&far), bulk("1"), bulk(src)], N) - .is_none(), - "ZDIFFSTORE is unimplemented; t2k4 owns the moment that changes" - ); + // ZDIFFSTORE used to be asserted here as OUT of the family, on the + // grounds that an unimplemented command has no write to misplace. + // moon#959 implemented it, so it moved INTO `FAMILY` above and is + // asserted positively there — the migration the `t2k4` tripwire in + // `tests/two_key_write_cross_shard.rs` existed to force. // A SORT with no STORE clause names one key: nothing to straddle. let sort_ro = [bulk(src), bulk("LIMIT"), bulk("0"), bulk("10")]; diff --git a/tests/two_key_write_cross_shard.rs b/tests/two_key_write_cross_shard.rs index ac3391303..6605ff4f6 100644 --- a/tests/two_key_write_cross_shard.rs +++ b/tests/two_key_write_cross_shard.rs @@ -214,6 +214,19 @@ const PROBES: &[Probe] = &[ src_untouched: ":2\r\n", src_after_success: None, }, + // moon#959 implemented ZDIFFSTORE, which is what moved it out of the + // t2k4 tripwire below and into the family. + Probe { + label: "ZDIFFSTORE", + seed: &[&["ZADD", "{s}", "1", "a", "2", "b"]], + argv: &["ZDIFFSTORE", "{d}", "1", "{s}"], + dst_probe: &["ZCARD", "{d}"], + dst_landed: ":2\r\n", + dst_absent: ":0\r\n", + src_probe: &["ZCARD", "{s}"], + src_untouched: ":2\r\n", + src_after_success: None, + }, Probe { label: "PFMERGE", seed: &[&["PFADD", "{s}", "a", "b", "c"]], @@ -549,23 +562,19 @@ fn t2k3_hash_tagged_pairs_still_work_at_four_shards() { ); } -/// Tripwire for the member of the family moon does not implement yet. -/// -/// `ZDIFFSTORE` is not in the dispatch table, so it cannot misplace a -/// destination today, which is the only reason it is absent from `PROBES` -/// and from the routing guard. +/// Tripwire for store forms moon does not implement (or must never accept). /// -/// If this test ever fails, it started working — and it went in WITHOUT a -/// cross-shard guard, which means it shipped the moon#592 defect. Add it to -/// `PROBES` and to `shared::touches_a_key_it_did_not_route_on`'s family list -/// in the same change. +/// If a row here ever answers a success, it started working — and it went in +/// WITHOUT a cross-shard guard, which means it shipped the moon#592 defect. +/// Add it to `PROBES` and to `shared::touches_a_key_it_did_not_route_on`'s +/// family list in the same change. /// -/// `GEORADIUS`/`GEORADIUSBYMEMBER ... STORE` were the other two rows here -/// until moon#645. They now work, so they are covered by `PROBES` above — -/// which is exactly the migration this tripwire exists to force. The `_RO` -/// twins still refuse the clause and are checked below, because a read-only -/// command that started writing would be a worse defect than the one this -/// file is about. +/// `GEORADIUS`/`GEORADIUSBYMEMBER ... STORE` were rows here until moon#645, +/// and `ZDIFFSTORE` until moon#959. They now work, so they are covered by +/// `PROBES` above — which is exactly the migration this tripwire exists to +/// force. The `_RO` twins still refuse the clause and are checked below, +/// because a read-only command that started writing would be a worse defect +/// than the one this file is about. #[test] fn t2k4_unimplemented_store_forms_stay_unimplemented_or_get_a_guard() { let m = spawn_moon(SHARDS); @@ -587,7 +596,6 @@ fn t2k4_unimplemented_store_forms_stay_unimplemented_or_get_a_guard() { assert_eq!(seeded, ":2\r\n", "GEOADD must seed {src}"); } let cases: &[(&str, &[&str])] = &[ - ("ZDIFFSTORE", &["ZDIFFSTORE", "t2k:zd:d", "1", "t2k:zd:s"]), ( "GEORADIUS_RO STORE", &[ diff --git a/tests/zset_read_cold_tier_928.rs b/tests/zset_read_cold_tier_928.rs index 2e64e353e..bceadf622 100644 --- a/tests/zset_read_cold_tier_928.rs +++ b/tests/zset_read_cold_tier_928.rs @@ -73,6 +73,8 @@ fn read(db: &mut Database, name: &str, args: &[&[u8]]) -> Frame { "ZREVRANGE" => sorted_set::zrevrange(db, &f), "ZRANGEBYSCORE" => sorted_set::zrangebyscore(db, &f), "ZREVRANGEBYSCORE" => sorted_set::zrevrangebyscore(db, &f), + "ZRANGEBYLEX" => sorted_set::zrangebylex(db, &f), + "ZREVRANGEBYLEX" => sorted_set::zrevrangebylex(db, &f), "ZCOUNT" => sorted_set::zcount(db, &f), "ZLEXCOUNT" => sorted_set::zlexcount(db, &f), "ZMSCORE" => sorted_set::zmscore(db, &f), @@ -102,6 +104,10 @@ fn reads() -> Vec<(&'static str, Vec<&'static [u8]>)> { vec![&b"z"[..], b"-inf", b"+inf", b"WITHSCORES"], ), ("ZREVRANGEBYSCORE", vec![&b"z"[..], b"+inf", b"-inf"]), + // moon#959: the two lex reads share the helpers above and must answer + // a cold zset the same way. + ("ZRANGEBYLEX", vec![&b"z"[..], b"-", b"+"]), + ("ZREVRANGEBYLEX", vec![&b"z"[..], b"+", b"[b"]), ("ZCOUNT", vec![&b"z"[..], b"(1", b"3"]), ("ZLEXCOUNT", vec![&b"z"[..], b"[b", b"+"]), ("ZMSCORE", vec![&b"z"[..], b"a", b"nope", b"c"]),