githubmirror/redis

src/t_array.c
2 021 строка71 KB
Salvatore Sanfilippo
Implement the new Redis Array type (#15162)
13 май 2026, 19:56
13 май 2026, 19:560d95764
/*
 * Copyright (c) 2026-Present, Redis Ltd.
 * All rights reserved.
 *
 * Licensed under your choice of (a) the Redis Source Available License 2.0
 * (RSALv2); or (b) the Server Side Public License v1 (SSPLv1); or (c) the
 * GNU Affero General Public License v3 (AGPLv3).
 *
 * Redis Array commands implementation.
 * Originally authored by: Salvatore Sanfilippo.
 *
 * The core sparse array data structure is in sparsearray.c/sparsearray.h.
 * This file contains Redis command handlers and Redis-specific operations.
 */

#include "server.h"
#include "../deps/tre/local_includes/tre.h"
#include <math.h>

/******************************************************************************
 *
 * ARRAY COMMANDS AND HIGHER LEVEL LOGIC
 *
 * This section contains all the Redis commands for the Array type, as well
 * as the type operations used by COPY and other server-level functionality.
 *
 *****************************************************************************/

/* ----------------------------------------------------------------------------
 * Array type operations for COPY command
 * -------------------------------------------------------------------------- */

robj *arrayTypeDup(robj *o) {
    redisArray *ar = o->ptr;
    redisArray *dup = arDup(ar);
    robj *newobj = createObject(OBJ_ARRAY, dup);
    newobj->encoding = OBJ_ENCODING_SLICED_ARRAY;
    return newobj;
}

/* ----------------------------------------------------------------------------
 * Internal helpers
 * -------------------------------------------------------------------------- */

#define ARGETRANGE_MAX_ITEMS 1000000

/* Lookup array object for write, create it if missing, or reply with
 * WRONGTYPE and return NULL if the key holds a different type. */
robj *lookupArrayForWriteOrReply(client *c, robj *key) {
    robj *o = lookupKeyWrite(c->db, key);
    if (o == NULL) {
        o = createArrayObject();
        dbAdd(c->db, key, &o);
    } else if (checkType(c, o, OBJ_ARRAY)) {
        return NULL;
    }
    return o;
}

/* Reply with an array value. This helper is needed because we used
 * tagged pointers for inlining values like floats, integers, small
 * strings directly inside the pointer. Big memory saves, but more
 * work needed when there is to reply to the client. */
void addReplyArrayValue(client *c, void *v) {
    if (arIsEmpty(v)) {
        addReplyNull(c);
        return;
    }

    char buf[AR_INLINE_BUFSIZE];
    size_t len;
    const char *data = arDecode(v, buf, sizeof(buf), &len);
    addReplyBulkCBuffer(c, data, len);
}

/* Parse array index from object. Accepts 0 to 2^64-2 by default.
 * If allow_max is true, also accepts UINT64_MAX. This is used by ARSEEK
 * because ARSEEK UINT64_MAX sets insert_idx to UINT64_MAX-1, which is
 * a valid terminal state (next ARINSERT would overflow).
 * Returns C_OK/C_ERR. Does NOT send error reply - caller decides. */
int getArrayIndexFromObject(robj *o, uint64_t *idx, int allow_max) {
    unsigned long long ull;
    if (o->encoding == OBJ_ENCODING_INT) {
        if ((long)o->ptr < 0) return C_ERR;
        ull = (unsigned long long)(long)o->ptr;
    } else {
        if (!string2ull(o->ptr, &ull)) return C_ERR;
    }
    if (ull == UINT64_MAX && !allow_max) return C_ERR;
    *idx = ull;
    return C_OK;
}

/* Parse an array index argument and reply with an error on failure. */
int arrayParseIndexOrReply(client *c, robj *arg, uint64_t *idx) {
    if (getArrayIndexFromObject(arg, idx, 0) != C_OK) {
        addReplyError(c, "invalid array index");
        return C_ERR;
    }
    return C_OK;
}

/* ----------------------------------------------------------------------------
 * ARGET / ARMGET
 * -------------------------------------------------------------------------- */

/* ARGET key idx
 *
 * Returns the value at idx in O(1).
 * Missing keys and holes both reply with NULL. */
void argetCommand(client *c) {
    robj *o = lookupKeyRead(c->db, c->argv[1]);
    if (o && checkType(c, o, OBJ_ARRAY)) return;

    uint64_t idx;
    if (arrayParseIndexOrReply(c, c->argv[2], &idx) != C_OK) return;

    void *v = o ? arGet(o->ptr, idx) : NULL;
    addReplyArrayValue(c, v);
}

/* ARMGET key idx [idx ...]
 *
 * Returns the values at the requested indices in O(N), where N is the number
 * of indices. Missing keys and holes reply with NULLs. All indices are
 * validated before the reply starts, so malformed input fails atomically. */
void armgetCommand(client *c) {
    robj *o = lookupKeyRead(c->db, c->argv[1]);
    if (o != NULL && checkType(c, o, OBJ_ARRAY)) return;

    /* Pre-validate all indices so malformed input fails the whole command,
     * like the other array commands. */
    for (int i = 2; i < c->argc; i++) {
        uint64_t idx;
        if (arrayParseIndexOrReply(c, c->argv[i], &idx) != C_OK) return;
    }

    addReplyArrayLen(c, c->argc - 2);

    for (int i = 2; i < c->argc; i++) {
        if (o == NULL) {
            /* Non existing keys are semantically equivalent
             * to non existing indexes of existing arrays. */
            addReplyNull(c);
            continue;
        }

        uint64_t idx = 0;
        getArrayIndexFromObject(c->argv[i], &idx, 0);  /* Already validated. */

        redisArray *ar = o->ptr;
        void *v = arGet(ar, idx);
        addReplyArrayValue(c, v);
    }
}

/* ----------------------------------------------------------------------------
 * ARSET / ARMSET
 * -------------------------------------------------------------------------- */

/* ARSET key <index> <value> [value ...]
 *
 * Sets one or more contiguous values in O(N), where N is the number of
 * values. Creates the array if needed and returns the number of previously
 * empty slots that were filled. */
 void arsetCommand(client *c) {
    uint64_t start_idx;
    if (arrayParseIndexOrReply(c, c->argv[2], &start_idx) != C_OK) return;

    int num_values = c->argc - 3;

    /* Pre-validate: check for overflow and forbidden max index. */
    uint64_t last_idx = start_idx + (uint64_t)num_values - 1;
    if (last_idx < start_idx || last_idx == UINT64_MAX) {
        addReplyError(c, "array index overflow");
        return;
    }

    robj *o = lookupArrayForWriteOrReply(c, c->argv[1]);
    if (o == NULL) return;

    redisArray *ar = o->ptr;
    uint64_t old_count = arCount(ar);
    size_t old_alloc = 0;
    if (server.memory_tracking_enabled) old_alloc = kvobjAllocSize(o);

    /* Pre-promote sparse slices only for true bulk sets. A single-element
     * write does not benefit from the extra range-analysis pass. */
    if (num_values > 1)
        arMayPromoteToDenseForRangeSet(ar, start_idx, last_idx);

    /* Write all values starting at start_idx */
    uint64_t idx = start_idx;
    for (int i = 3; i < c->argc; i++) {
        sds val = c->argv[i]->ptr;
        void *v = arEncode(val, sdslen(val));
        arSet(ar, idx, v);
        idx++;
    }

    long long set_count = arCount(ar) - old_count;
    updateKeysizesHist(c->db, OBJ_ARRAY, old_count, arCount(ar));
    if (server.memory_tracking_enabled)
        updateSlotAllocSize(c->db, getKeySlot(c->argv[1]->ptr), o, old_alloc, kvobjAllocSize(o));
    keyModified(c, c->db, c->argv[1], o, 1);
    notifyKeyspaceEvent(NOTIFY_ARRAY, "arset", c->argv[1], c->db->id);
    server.dirty += num_values;
    addReplyLongLong(c, set_count);
}

/* ARMSET key idx value [idx value ...]
 *
 * Sets multiple scattered index/value pairs in O(N), where N is the number of
 * pairs. Creates the array if needed, returns the number of newly filled
 * slots, and validates all indices before mutating. */
void armsetCommand(client *c) {
    if ((c->argc - 2) % 2 != 0) {
        addReplyErrorArity(c);
        return;
    }

    /* Validate all indices first */
    for (int i = 2; i < c->argc; i += 2) {
        uint64_t idx;
        if (arrayParseIndexOrReply(c, c->argv[i], &idx) != C_OK) return;
    }

    robj *o = lookupArrayForWriteOrReply(c, c->argv[1]);
    if (o == NULL) return;

    redisArray *ar = o->ptr;
    uint64_t old_count = arCount(ar);
    size_t old_alloc = 0;
    if (server.memory_tracking_enabled) old_alloc = kvobjAllocSize(o);

    for (int i = 2; i < c->argc; i += 2) {
        uint64_t idx = 0;
        getArrayIndexFromObject(c->argv[i], &idx, 0);  /* Already validated */

        sds val = c->argv[i + 1]->ptr;
        void *v = arEncode(val, sdslen(val));
        arSet(ar, idx, v);
    }

    int num_pairs = (c->argc - 2) / 2;
    long long set_count = arCount(ar) - old_count;
    updateKeysizesHist(c->db, OBJ_ARRAY, old_count, arCount(ar));
    if (server.memory_tracking_enabled)
        updateSlotAllocSize(c->db, getKeySlot(c->argv[1]->ptr), o, old_alloc, kvobjAllocSize(o));
    keyModified(c, c->db, c->argv[1], o, 1);
    notifyKeyspaceEvent(NOTIFY_ARRAY, "armset", c->argv[1], c->db->id);
    server.dirty += num_pairs;
    addReplyLongLong(c, set_count);
}

/* ----------------------------------------------------------------------------
 * ARDEL / ARDELRANGE
 * -------------------------------------------------------------------------- */

/* ARDEL key idx [idx ...]
 *
 * Deletes the specified indices in O(N), where N is the number of indices.
 * All indices are validated first, and if the array becomes empty the key
 * itself is deleted. The number of deleted (existing) items is returned. */
void ardelCommand(client *c) {
    /* Pre-validate all indices before mutating, to report syntax errors
     * even if the key doesn't exist. */
    for (int i = 2; i < c->argc; i++) {
        uint64_t idx;
        if (arrayParseIndexOrReply(c, c->argv[i], &idx) != C_OK) return;
    }

    robj *o = lookupKeyWrite(c->db, c->argv[1]);
    if (o == NULL) {
        addReplyLongLong(c, 0);
        return;
    }
    if (checkType(c, o, OBJ_ARRAY)) return;

    redisArray *ar = o->ptr;
    uint64_t old_count = arCount(ar);
    size_t old_alloc = 0;
    if (server.memory_tracking_enabled) old_alloc = kvobjAllocSize(o);
    long long deleted = 0;

    for (int i = 2; i < c->argc; i++) {
        uint64_t idx = 0;
        getArrayIndexFromObject(c->argv[i], &idx, 0);  /* Already validated */
        deleted += arDel(ar, idx);
    }

    int keyremoved = (arCount(ar) == 0);
    if (server.memory_tracking_enabled && deleted > 0 && keyremoved)
        updateSlotAllocSize(c->db, getKeySlot(c->argv[1]->ptr), o, old_alloc, kvobjAllocSize(o));
    if (deleted > 0) {
        if (keyremoved)
            dbDeleteSkipKeysizesUpdate(c->db, c->argv[1]);
        updateKeysizesHist(c->db, OBJ_ARRAY,
                           old_count, keyremoved ? -1 : (int64_t)arCount(ar));
        if (server.memory_tracking_enabled && !keyremoved)
            updateSlotAllocSize(c->db, getKeySlot(c->argv[1]->ptr), o, old_alloc, kvobjAllocSize(o));
        keyModified(c, c->db, c->argv[1], keyremoved ? NULL : o, 1);
        notifyKeyspaceEvent(NOTIFY_ARRAY, "ardel", c->argv[1], c->db->id);
        if (keyremoved)
            notifyKeyspaceEvent(NOTIFY_GENERIC, "del", c->argv[1], c->db->id);
        server.dirty += deleted;
    }
    addReplyLongLong(c, deleted);
}

/* ARDELRANGE key start end [start end ...]
 *
 * Deletes one or more ranges. Complexity is proportional to the existing
 * elements / slices touched, not to the numeric span of the requested ranges,
 * so huge ranges do not block the server forever.
 *
 * Each pair may be given in either order. All ranges are validated up front,
 * and an empty resulting array deletes the key. */
void ardelrangeCommand(client *c) {
    if ((c->argc - 2) % 2 != 0) {
        addReplyErrorArity(c);
        return;
    }

    /* Pre-validate all ranges before mutating, to avoid partial updates
     * if a later range has invalid syntax. */
    for (int i = 2; i < c->argc; i += 2) {
        uint64_t start, end;
        if (arrayParseIndexOrReply(c, c->argv[i], &start) != C_OK) return;
        if (arrayParseIndexOrReply(c, c->argv[i + 1], &end) != C_OK) return;
    }

    robj *o = lookupKeyWrite(c->db, c->argv[1]);
    if (o == NULL) {
        addReplyLongLong(c, 0);
        return;
    }
    if (checkType(c, o, OBJ_ARRAY)) return;

    redisArray *ar = o->ptr;
    uint64_t old_count = arCount(ar);
    size_t old_alloc = 0;
    if (server.memory_tracking_enabled) old_alloc = kvobjAllocSize(o);
    uint64_t total_deleted = 0;

    /* Process each range using the generalized arDeleteRange */
    for (int i = 2; i < c->argc; i += 2) {
        uint64_t start = 0, end = 0;
        getArrayIndexFromObject(c->argv[i], &start, 0); /* Already validated */
        getArrayIndexFromObject(c->argv[i + 1], &end, 0);

        uint64_t lo = (start <= end) ? start : end;
        uint64_t hi = (start <= end) ? end : start;

        total_deleted += arDeleteRange(ar, lo, hi);
    }

    int keyremoved = (arCount(ar) == 0);
    if (server.memory_tracking_enabled && total_deleted > 0 && keyremoved)
        updateSlotAllocSize(c->db, getKeySlot(c->argv[1]->ptr), o, old_alloc, kvobjAllocSize(o));
    if (total_deleted > 0) {
        if (keyremoved)
            dbDeleteSkipKeysizesUpdate(c->db, c->argv[1]);
        updateKeysizesHist(c->db, OBJ_ARRAY,
                           old_count, keyremoved ? -1 : (int64_t)arCount(ar));
        if (server.memory_tracking_enabled && !keyremoved)
            updateSlotAllocSize(c->db, getKeySlot(c->argv[1]->ptr), o, old_alloc, kvobjAllocSize(o));
        keyModified(c, c->db, c->argv[1], keyremoved ? NULL : o, 1);
        notifyKeyspaceEvent(NOTIFY_ARRAY, "ardelrange", c->argv[1], c->db->id);
        if (keyremoved)
            notifyKeyspaceEvent(NOTIFY_GENERIC, "del", c->argv[1], c->db->id);
        server.dirty += total_deleted;
    }
    addReplyUnsignedLongLong(c, total_deleted);
}

/* ----------------------------------------------------------------------------
 * ARLEN / ARCOUNT
 * -------------------------------------------------------------------------- */

/* ARLEN key
 *
 * Returns max-index-plus-one in O(1).
 * Missing keys reply with 0. */
void arlenCommand(client *c) {
    robj *o = lookupKeyReadOrReply(c, c->argv[1], shared.czero);
    if (o == NULL || checkType(c, o, OBJ_ARRAY)) return;

    redisArray *ar = o->ptr;
    addReplyUnsignedLongLong(c, arLen(ar));
}

/* ARCOUNT key
 *
 * Returns the number of non-empty elements in O(1).
 * Missing keys reply with 0. */
void arcountCommand(client *c) {
    robj *o = lookupKeyReadOrReply(c, c->argv[1], shared.czero);
    if (o == NULL || checkType(c, o, OBJ_ARRAY)) return;

    redisArray *ar = o->ptr;
    addReplyUnsignedLongLong(c, arCount(ar));
}

/* ----------------------------------------------------------------------------
 * ARGETRANGE
 * -------------------------------------------------------------------------- */

/* ARGETRANGE key start end
 *
 * Returns every position in the requested range in O(N), where N is the range
 * length. Holes are returned as NULLs, and a missing key behaves like an all-
 * NULL array. If start > end the reply order is reversed.
 *
 * To avoid giant synthetic NULL replies, the range length is hard-limited,
 * otherwise the command, with a wrong range, could make the server totally
 * unusable. The max range is 1 million elements and is fixed, constituting
 * the user-facing semantic of the command. */
void argetrangeCommand(client *c) {
    uint64_t start, end;
    if (arrayParseIndexOrReply(c, c->argv[2], &start) != C_OK) return;
    if (arrayParseIndexOrReply(c, c->argv[3], &end) != C_OK) return;

    robj *o = lookupKeyRead(c->db, c->argv[1]);
    if (o != NULL && checkType(c, o, OBJ_ARRAY)) return;

    int reverse = start > end;
    uint64_t lo = reverse ? end : start;
    uint64_t hi = reverse ? start : end;
    uint64_t len = hi - lo + 1;

    /* ARGETRANGE is a special command: it can trigger a huge reply blocking
     * the server (basically forever) even if there is no actual data. This
     * is unlike an SMEMBERS against a very big key: somebody added so many
     * elements inside, before asking for a huge amount of elements. But, in the
     * case of ARGETRANGE, you can just trigger a huge amount of NULLs to be
     * sent to the client. In ARSCAN this was optimized to be O(N) with the
     * actual populated elements, but in this case it can't be done because
     * of the semantic of the command, and the Redis protocol inability to reply
     * with run-length ranges (a, b, c, 1293455 NULLs, d, e).
     *
     * Because of all that, we put an hard limit in the range size, and this
     * limit must be part of the Redis culture, so it should not be tuned in
     * any way: 1 million items, with an hard error if the range is bigger than
     * that, not just a silent trimming at this length, that would cause hard
     * to track bugs. */
    if (len > ARGETRANGE_MAX_ITEMS) {
        addReplyErrorFormat(c, "range exceeds maximum of %u items",
            ARGETRANGE_MAX_ITEMS);
        return;
    }

    addReplyArrayLen(c, len);
    if (o == NULL) {
        for (uint64_t i = 0; i < len; i++) addReplyNull(c);
        return;
    }

    redisArray *ar = o->ptr;
    if (reverse) {
        for (uint64_t idx = hi; ; idx--) {
            void *v = arGet(ar, idx);
            addReplyArrayValue(c, v);
            if (idx == lo) break;
        }
    } else {
        for (uint64_t idx = lo; idx <= hi; idx++) {
            void *v = arGet(ar, idx);
            addReplyArrayValue(c, v);
        }
    }
}

/* ----------------------------------------------------------------------------
 * ARSCAN
 * -------------------------------------------------------------------------- */

/* Iterate populated elements in [start..end].
 *
 * This iterator is read-only and not mutation-stable: between Init() and the
 * final Next() that returns 0, the caller must not write to the array. Any
 * write may free or relocate the current slice, making the iterator state
 * stale. The goal of this abstraction was to capture repeated code in the
 * implementation of ARSCAN, ARGREP, AROP.
 *
 * The struct lives on the caller stack, so setup and iteration stay allocation
 * free and command-local. */
typedef struct {
    redisArray *ar;
    uint64_t lo;              /* Normalized inclusive range start. */
    uint64_t hi;              /* Normalized inclusive range end. */
    uint64_t lo_slice;        /* First slice touched by the range. */
    uint64_t hi_slice;        /* Last slice touched by the range. */
    uint32_t slice_size;      /* Cached slice size. */
    int reverse;              /* Iterate from high to low. */
    int32_t step;             /* +1 forward, -1 backward. */
    int done;                 /* No more elements to return. */
    int top_done;             /* No more slices to inspect after current. */

    uint64_t slice_id;        /* Next flat-directory slice to inspect. */
    int32_t sdir_index;       /* Next superdir entry to inspect. */
    int32_t slot_index;       /* Next slot inside the current superdir entry. */

    arSlice *slice;           /* Slice currently being scanned. */
    uint64_t slice_base;      /* Logical index of slice offset 0. */
    uint32_t off_lo;          /* First in-range offset for current slice. */
    uint32_t off_hi;          /* Last in-range offset for current slice. */
    int dense;                /* Current slice is dense. */
    void **dense_items;       /* Dense items window. */
    int32_t dense_off;        /* Current dense logical offset. */
    int32_t dense_item_pos;   /* Current dense window index. */
    int32_t dense_item_end;   /* Final dense window index. */
    uint16_t *sparse_offsets; /* Sparse offsets array. */
    void **sparse_values;     /* Sparse values array. */
    int32_t sparse_count;     /* Sparse entry count. */
    int32_t sparse_pos;       /* Current sparse entry position. */
    int slice_ready;          /* Current slice scan state is initialized. */
} arScanIter;

#define AR_SCAN_ITER_SLOT_UNSET INT32_MIN

/* Keep the per-element iterator hot path inline in the command loops.
 * It helps a lot with certain targets, up to ~30-50% speed regression
 * without forcing the inlining. */
#if defined(__GNUC__) || defined(__clang__)
#define ALWAYS_INLINE __attribute__((always_inline)) inline
#else
#define ALWAYS_INLINE inline
#endif

/* Initialize a populated-elements iterator. Empty arrays and empty clipped
 * ranges are turned into a done iterator here so the first Next() is a single
 * branch. */
static void arScanIterInit(redisArray *ar, uint64_t start, uint64_t end,
                           arScanIter *it)
{
    memset(it, 0, sizeof(*it));
    it->ar = ar;

    if (ar == NULL || arCount(ar) == 0) {
        it->done = 1;
        it->top_done = 1;
        return;
    }

    /* Note that a few things here could be taken
     * from the array itself, as they are immutable,
     * but after introducing this abstraction a small
     * but measurable speed regression suggested to
     * micro-optimize for this hot path and have
     * iterator-side copies of often used stuff. */
    it->reverse = start > end;
    it->step = it->reverse ? -1 : 1;
    it->lo = it->reverse ? end : start;
    it->hi = it->reverse ? start : end;
    it->slice_size = ar->slice_size;
    it->lo_slice = it->lo / it->slice_size;
    it->hi_slice = it->hi / it->slice_size;
    it->slot_index = AR_SCAN_ITER_SLOT_UNSET;

    /* No intersection between the range and the array span. */
    if (it->lo_slice > ar->dir_highest_used) {
        it->done = 1;
        it->top_done = 1;
        return;
    }

    /* Clip the high end to the actual array span. */
    if (it->hi_slice > ar->dir_highest_used) {
        it->hi_slice = ar->dir_highest_used;
        it->hi = arMakeIdx(it->hi_slice, it->slice_size - 1, it->slice_size);
    }

    /* Clipping made the range empty? */
    if (it->lo_slice > it->hi_slice) {
        it->done = 1;
        it->top_done = 1;
        return;
    }

    if (ar->superdir) {
        int found;

        /* Start from the first superdir block that can intersect the range. */
        uint64_t block_id = (it->reverse ? it->hi_slice : it->lo_slice) /
                            AR_SUPER_BLOCK_SLOTS;
        uint32_t pos = arSuperDirFind(ar, block_id, &found);

        if (it->reverse) it->sdir_index = found ? (int32_t)pos : (int32_t)pos - 1;
        else it->sdir_index = (int32_t)pos;

        /* No superdir block intersects the clipped range. */
        if (it->sdir_index < 0 || it->sdir_index >= (int32_t)ar->sdir_len) {
            it->done = 1;
            it->top_done = 1;
        }
    } else {
        /* Flat directory iteration starts directly from the first in-range slice. */
        it->slice_id = it->reverse ? it->hi_slice : it->lo_slice;
    }
}

/* Prepare the current slice-local scan state. Returns 1 if the slice may
 * yield at least one populated element in range, otherwise 0.
 * The function is used by arScanIterLoadNextSlice() each time a new
 * slice should be iterated. When a new slice is selected by
 * arScanIterLoadNextSlice(), then this function is called to setup the
 * iteration needed by arScanIterNext(). */
static ALWAYS_INLINE int arScanIterPrepareSlice(arScanIter *it,
                                                arSlice *s, uint64_t slice_id)
{
    uint64_t slice_base = slice_id * it->slice_size;
    /* Restrict the scan to the part of this slice touched by the query. */
    uint32_t off_lo = (slice_id == it->lo_slice) ?
        arSliceOff(it->lo, it->slice_size) : 0;
    uint32_t off_hi = (slice_id == it->hi_slice) ?
        arSliceOff(it->hi, it->slice_size) : it->slice_size - 1;

    if (s->encoding == AR_SLICE_DENSE) {
        uint32_t win_lo = s->layout.dense.offset;
        uint32_t win_hi = s->layout.dense.offset + s->layout.dense.winsize - 1;

        /* Dense slices may only have a smaller populated window allocated. */
        if (off_lo < win_lo) off_lo = win_lo;
        if (off_hi > win_hi) off_hi = win_hi;

        /* No intersection between the range and the dense window. */
        if (off_lo > off_hi) return 0;

        it->dense = 1;
        it->dense_items = s->layout.dense.items;
        it->dense_off = it->reverse ? (int32_t)off_hi : (int32_t)off_lo;
        it->dense_item_pos = it->dense_off - (int32_t)win_lo;
        it->dense_item_end = (it->reverse ? (int32_t)off_lo :
                              (int32_t)off_hi) - (int32_t)win_lo;
    } else {
        int found;
        uint32_t pos;
        uint16_t *offsets = s->layout.sparse.offsets;

        it->dense = 0;
        it->sparse_offsets = offsets;
        it->sparse_values = s->layout.sparse.values;
        it->sparse_count = (int32_t)s->count;
        if (it->reverse) {
            /* Start from the last sparse entry that can still be in range. */
            pos = arSparseFindPos(s, (uint16_t)off_hi, &found);
            it->sparse_pos = found ? (int32_t)pos : (int32_t)pos - 1;

            /* No sparse entry falls inside the requested offsets. */
            if (it->sparse_pos < 0 || offsets[it->sparse_pos] < off_lo)
                return 0;
        } else {
            /* Start from the first sparse entry that can still be in range. */
            pos = arSparseFindPos(s, (uint16_t)off_lo, &found);
            it->sparse_pos = (int32_t)pos;

            /* No sparse entry falls inside the requested offsets. */
            if (it->sparse_pos >= (int32_t)s->count ||
                offsets[it->sparse_pos] > off_hi) return 0;
        }
    }

    it->slice = s;
    it->slice_base = slice_base;
    it->off_lo = off_lo;
    it->off_hi = off_hi;
    it->slice_ready = 1;
    return 1;
}

/* Advance top-level directory state until a non-NULL slice in range is ready
 * for local scanning, or return 0 if the iterator is exhausted. */
static ALWAYS_INLINE int arScanIterLoadNextSlice(arScanIter *it) {
    redisArray *ar = it->ar;

    if (ar->superdir) {
        while (!it->top_done) {
            /* No more superdir blocks to inspect. */
            if (it->sdir_index < 0 || it->sdir_index >= (int32_t)ar->sdir_len) {
                it->top_done = 1;
                break;
            }

            arSDirEntry *e = ar->superdir + it->sdir_index;
            uint64_t block_base = e->block_id * AR_SUPER_BLOCK_SLOTS;
            uint64_t block_end = block_base + AR_SUPER_BLOCK_SLOTS - 1;
            int32_t block_slot_lo = (block_base < it->lo_slice) ?
                (int32_t)(it->lo_slice - block_base) : 0;
            int32_t block_slot_hi = (block_end > it->hi_slice) ?
                (int32_t)(it->hi_slice - block_base) : AR_SUPER_BLOCK_SLOTS - 1;

            /* This block starts after the requested range. */
            if (block_base > it->hi_slice) {
                it->top_done = 1;
                break;
            }

            /* This block ends before the requested range. */
            if (block_end < it->lo_slice) {
                if (it->reverse) it->top_done = 1;
                else it->sdir_index++;
                it->slot_index = AR_SCAN_ITER_SLOT_UNSET;
                continue;
            }

            if (it->reverse) {
                /* slot_index uses a sentinel outside the valid 0..2047 range
                 * so reverse scans can consume slot 0 and then fall below the
                 * block without looking like a fresh block entry. */
                if (it->slot_index == AR_SCAN_ITER_SLOT_UNSET)
                    it->slot_index = block_slot_hi;

                while (it->slot_index >= block_slot_lo) {
                    int32_t si = it->slot_index--;
                    arSlice *s = e->slots[si];
                    if (s && arScanIterPrepareSlice(it, s, block_base + si))
                        return 1;
                }

                /* This block had no more matching slices, move to the previous block. */
                it->sdir_index--;
                it->slot_index = AR_SCAN_ITER_SLOT_UNSET;
            } else {
                /* slot_index uses a sentinel outside the valid 0..2047 range
                 * so an exhausted block does not look like a fresh entry. */
                if (it->slot_index == AR_SCAN_ITER_SLOT_UNSET)
                    it->slot_index = block_slot_lo;

                while (it->slot_index <= block_slot_hi) {
                    int32_t si = it->slot_index++;
                    arSlice *s = e->slots[si];
                    if (s && arScanIterPrepareSlice(it, s, block_base + si))
                        return 1;
                }

                /* This block had no more matching slices, move to the next block. */
                it->sdir_index++;
                it->slot_index = AR_SCAN_ITER_SLOT_UNSET;
            }
        }
    } else {
        while (!it->top_done) {
            uint64_t slice_id = it->slice_id;
            arSlice *s = ar->dir[slice_id];

            /* Advance the top-level cursor before possibly returning this slice. */
            if (it->reverse) {
                if (slice_id == it->lo_slice) it->top_done = 1;
                else it->slice_id = slice_id - 1;
            } else {
                if (slice_id == it->hi_slice) it->top_done = 1;
                else it->slice_id = slice_id + 1;
            }

            if (s && arScanIterPrepareSlice(it, s, slice_id))
                return 1;
        }
    }

    return 0;
}

/* Return the next populated element in range, or 0 when done. */
static ALWAYS_INLINE int arScanIterNext(arScanIter *it,
                                        uint64_t *idx, void **value)
{
    /* The iterator was already fully consumed. */
    if (it->done) return 0;

    while (1) {
        if (it->slice_ready) {
            /* Drain the current slice before asking for another one. */
            if (it->dense) {
                while ((it->step > 0 && it->dense_item_pos <= it->dense_item_end) ||
                       (it->step < 0 && it->dense_item_pos >= it->dense_item_end)) {
                    uint32_t off = (uint32_t)it->dense_off;
                    void *v = it->dense_items[it->dense_item_pos];
                    it->dense_off += it->step;
                    it->dense_item_pos += it->step;

                    /* Dense windows may contain holes. */
                    if (arIsEmpty(v)) continue;

                    if (idx) *idx = it->slice_base + off;
                    *value = v;
                    return 1;
                }
            } else {
                while (it->sparse_pos >= 0 && it->sparse_pos < it->sparse_count) {
                    int32_t pos = it->sparse_pos;
                    uint32_t off = it->sparse_offsets[pos];

                    /* Sparse entries are sorted, so leaving the window ends this slice. */
                    if (off < it->off_lo || off > it->off_hi) break;

                    it->sparse_pos += it->step;
                    if (idx) *idx = it->slice_base + off;
                    *value = it->sparse_values[pos];
                    return 1;
                }
            }

            /* The current slice has no more in-range populated elements. */
            it->slice = NULL;
            it->slice_ready = 0;
        }

        /* No more in-range slices are available. */
        if (!arScanIterLoadNextSlice(it)) {
            it->done = 1;
            return 0;
        }
    }
}

/* ARSCAN key start end [LIMIT count]
 *
 * Returns only existing elements as flat index/value pairs.
 *
 * Complexity is O(P), where P is visited positions in touched slices
 * (dense scanned slots + sparse entries), with worst-case O(|end-start|+1)
 * and typical case close to O(N), where N is the number of existing
 * elements in range. This means that huge ranges are safe and will not
 * block the server with a work bound to the span length.
 *
 * Unlike ARGETRANGE, holes are skipped rather than returned as NULLs.
 * LIMIT caps the number of returned pairs. */
void arscanCommand(client *c) {
    uint64_t start, end;
    if (arrayParseIndexOrReply(c, c->argv[2], &start) != C_OK) return;
    if (arrayParseIndexOrReply(c, c->argv[3], &end) != C_OK) return;

    /* Parse optional LIMIT */
    uint64_t remaining = UINT64_MAX;
    if (c->argc == 6) {
        if (strcasecmp(c->argv[4]->ptr, "LIMIT") != 0) {
            addReplyErrorObject(c, shared.syntaxerr);
            return;
        }
        long long ll;
        if (getLongLongFromObjectOrReply(c, c->argv[5], &ll, NULL) != C_OK)
            return;
        if (ll <= 0) {
            addReplyError(c, "LIMIT must be positive");
            return;
        }
        remaining = (uint64_t)ll;
    } else if (c->argc != 4) {
        addReplyErrorArity(c);
        return;
    }

    robj *o = lookupKeyRead(c->db, c->argv[1]);
    if (o != NULL && checkType(c, o, OBJ_ARRAY)) return;

    if (o == NULL) {
        addReplyArrayLen(c, 0);
        return;
    }

    redisArray *ar = o->ptr;
    void *replylen = addReplyDeferredLen(c);
    uint64_t count = 0;
    arScanIter it;
    uint64_t idx;
    void *v;

    arScanIterInit(ar, start, end, &it);
    while (remaining && arScanIterNext(&it, &idx, &v)) {
        /* Reply with nested [idx, value] pairs. */
        addReplyArrayLen(c, 2);
        addReplyUnsignedLongLong(c, idx);
        addReplyArrayValue(c, v);
        count++;
        remaining--;
    }

    setDeferredArrayLen(c, replylen, count);
}

/* ============================================================================
 * ARGREP
 * ============================================================================
 *
 * Search existing array elements in a range using textual predicates.
 * Like ARSCAN, the work is bound by the visited slices, not by the raw
 * numeric span alone: dense slices scan the touched dense window, while
 * sparse slices only scan stored entries inside the covered offsets.
 * -------------------------------------------------------------------------- */

#define ARGREP_PRED_EXACT 1
#define ARGREP_PRED_MATCH 2
#define ARGREP_PRED_GLOB  3
#define ARGREP_PRED_RE    4

#define ARGREP_MAX_PREDICATES 250
#define ARGREP_MAX_RE_LEN 2048

#define ARGREP_COMBINE_OR  1
#define ARGREP_COMBINE_AND 2

#define ARGREP_BOUND_INDEX 1
#define ARGREP_BOUND_START 2
#define ARGREP_BOUND_END   3

typedef struct {
    int type;               /* EXACT, MATCH, GLOB, or RE. */
    sds pattern;            /* Pattern argument exactly as given by the user. */
    regex_t regex;          /* Compiled regex for RE predicates. */
    int regex_compiled;     /* Whether regex must be freed. */
} arGrepPredicate;

typedef struct {
    int type;               /* Numeric index, logical start, or logical end. */
    uint64_t index;         /* Used only for numeric bounds. */
} arGrepBound;

typedef struct {
    arGrepPredicate *preds; /* All predicates to apply to each element. */
    int num_preds;          /* Number of predicates stored in preds[]. */
    int combine;            /* OR by default, AND if requested. */
    int withvalues;         /* Reply with [idx value ...] instead of [idx ...]. */
    int nocase;             /* Apply case-insensitive matching globally. */
} arGrepPlan;

/* Lowercase only ASCII letters. This keeps MATCH/EXACT deterministic and
 * locale-independent even on arbitrary binary payloads. */
static inline unsigned char arGrepLowerAscii(unsigned char c) {
    return (c >= 'A' && c <= 'Z') ? (unsigned char)(c + ('a' - 'A')) : c;
}

/* Compare two byte strings, optionally ignoring ASCII case. */
int arGrepBytesEqual(const char *a, size_t alen, const char *b, size_t blen,
                     int nocase) {
    if (alen != blen) return 0;
    if (!nocase) return memcmp(a, b, alen) == 0;

    for (size_t i = 0; i < alen; i++) {
        if (arGrepLowerAscii((unsigned char)a[i]) !=
            arGrepLowerAscii((unsigned char)b[i])) {
            return 0;
        }
    }
    return 1;
}

/* Find a needle inside a byte string, optionally ignoring ASCII case. */
int arGrepBytesContains(const char *haystack, size_t haystack_len,
                        const char *needle, size_t needle_len, int nocase) {
    if (needle_len == 0) return 1;
    if (needle_len > haystack_len) return 0;

    size_t last = haystack_len - needle_len;
    for (size_t i = 0; i <= last; i++) {
        if (arGrepBytesEqual(haystack + i, needle_len, needle, needle_len,
                             nocase)) {
            return 1;
        }
    }
    return 0;
}

/* Return the predicate type for a keyword, or 0 if it is not one. */
int arGrepPredicateType(const char *token) {
    if (!strcasecmp(token, "EXACT")) return ARGREP_PRED_EXACT;
    if (!strcasecmp(token, "MATCH")) return ARGREP_PRED_MATCH;
    if (!strcasecmp(token, "GLOB")) return ARGREP_PRED_GLOB;
    if (!strcasecmp(token, "RE")) return ARGREP_PRED_RE;
    return 0;
}

/* Free any compiled regex state created while parsing ARGREP. */
void arGrepFreePlan(arGrepPlan *plan) {
    if (plan->preds == NULL) return;

    for (int i = 0; i < plan->num_preds; i++) {
        if (plan->preds[i].regex_compiled)
            tre_regfree(&plan->preds[i].regex);
    }
    zfree(plan->preds);
    plan->preds = NULL;
}

/* Parse a bound argument. ARGREP accepts the special tokens "-" and "+"
 * in addition to normal array indexes. */
int arGrepParseBoundOrReply(client *c, robj *arg, arGrepBound *bound) {
    if (arg->encoding != OBJ_ENCODING_INT) {
        sds token = arg->ptr;
        if (sdslen(token) == 1 && token[0] == '-') {
            bound->type = ARGREP_BOUND_START;
            bound->index = 0;
            return C_OK;
        }
        if (sdslen(token) == 1 && token[0] == '+') {
            bound->type = ARGREP_BOUND_END;
            bound->index = 0;
            return C_OK;
        }
    }

    if (getArrayIndexFromObject(arg, &bound->index, 0) != C_OK) {
        addReplyError(c, "invalid array index");
        return C_ERR;
    }
    bound->type = ARGREP_BOUND_INDEX;
    return C_OK;
}

/* Resolve a parsed bound against the current array length. */
uint64_t arGrepResolveBound(arGrepBound *bound, uint64_t max_index) {
    if (bound->type == ARGREP_BOUND_START) return 0;
    if (bound->type == ARGREP_BOUND_END) return max_index;
    return bound->index;
}

/* Compile all RE predicates after the whole command is parsed, so NOCASE is
 * already known and affects every regex consistently. */
int arGrepCompileRegexesOrReply(client *c, arGrepPlan *plan) {
    for (int i = 0; i < plan->num_preds; i++) {
        arGrepPredicate *pred = &plan->preds[i];
        if (pred->type != ARGREP_PRED_RE) continue;

        if (sdslen(pred->pattern) == 0) {
            addReplyError(c, "regular expression is empty");
            return C_ERR;
        }

        int cflags = REG_EXTENDED | REG_NOSUB | REG_USEBYTES;
        if (plan->nocase) cflags |= REG_ICASE;

        int err = tre_regncompb(&pred->regex, pred->pattern,
                                sdslen(pred->pattern), cflags);
        if (err != REG_OK) {
            char errbuf[256];
            tre_regerror(err, &pred->regex, errbuf, sizeof(errbuf));
            addReplyErrorFormat(c, "invalid regular expression: %s", errbuf);
            return C_ERR;
        }
        pred->regex_compiled = 1;

        if (tre_have_backrefs(&pred->regex)) {
            addReplyError(c, "regular expression backreferences are not supported");
            return C_ERR;
        }
    }
    return C_OK;
}

/* Parse predicates and global modifiers in a single pass. This makes the
 * command more user-friendly because predicates and options can be mixed
 * freely. If the same global option appears multiple times, the last one
 * wins. */
int arGrepParsePlanOrReply(client *c, arGrepPlan *plan, uint64_t *limit) {
    memset(plan, 0, sizeof(*plan));
    plan->combine = ARGREP_COMBINE_OR;
    *limit = UINT64_MAX;

    int max_preds = c->argc - 4;
    plan->preds = zcalloc(sizeof(*plan->preds) * max_preds);

    for (int arg = 4; arg < c->argc; ) {
        sds token = c->argv[arg]->ptr;
        int type = arGrepPredicateType(token);

        if (type != 0) {
            if (arg + 1 >= c->argc) {
                addReplyErrorObject(c, shared.syntaxerr);
                return C_ERR;
            }
            if (plan->num_preds >= ARGREP_MAX_PREDICATES) {
                addReplyErrorFormat(c, "too many predicates, maximum is %d",
                                    ARGREP_MAX_PREDICATES);
                return C_ERR;
            }

            arGrepPredicate *pred = &plan->preds[plan->num_preds++];
            pred->type = type;
            pred->pattern = c->argv[arg + 1]->ptr;
            if (type == ARGREP_PRED_RE &&
                sdslen(pred->pattern) > ARGREP_MAX_RE_LEN) {
                addReplyErrorFormat(c,
                    "regular expression is too long, maximum is %d bytes",
                    ARGREP_MAX_RE_LEN);
                return C_ERR;
            }
            arg += 2;
            continue;
        }

        if (!strcasecmp(token, "LIMIT")) {
            if (arg + 1 >= c->argc) {
                addReplyErrorObject(c, shared.syntaxerr);
                return C_ERR;
            }

            long long ll;
            if (getLongLongFromObjectOrReply(c, c->argv[arg + 1], &ll, NULL)
                != C_OK) {
                return C_ERR;
            }
            if (ll <= 0) {
                addReplyError(c, "LIMIT must be positive");
                return C_ERR;
            }

            *limit = (uint64_t)ll;
            arg += 2;
            continue;
        }

        if (!strcasecmp(token, "WITHVALUES")) {
            plan->withvalues = 1;
            arg++;
            continue;
        }

        if (!strcasecmp(token, "NOCASE")) {
            plan->nocase = 1;
            arg++;
            continue;
        }

        if (!strcasecmp(token, "AND") || !strcasecmp(token, "OR")) {
            plan->combine = !strcasecmp(token, "AND") ?
                ARGREP_COMBINE_AND : ARGREP_COMBINE_OR;
            arg++;
            continue;
        }

        addReplyErrorObject(c, shared.syntaxerr);
        return C_ERR;
    }

    if (plan->num_preds == 0) {
        addReplyErrorObject(c, shared.syntaxerr);
        return C_ERR;
    }

    return arGrepCompileRegexesOrReply(c, plan);
}

/* Match one predicate against the decoded element bytes. */
int arGrepMatchPredicate(arGrepPredicate *pred, const char *data, size_t len,
                         int nocase) {
    size_t pattern_len = sdslen(pred->pattern);

    switch (pred->type) {
    case ARGREP_PRED_EXACT:
        return arGrepBytesEqual(data, len, pred->pattern, pattern_len, nocase);
    case ARGREP_PRED_MATCH:
        return arGrepBytesContains(data, len, pred->pattern, pattern_len,
                                   nocase);
    case ARGREP_PRED_GLOB:
        return stringmatchlen(pred->pattern, pattern_len, data, len, nocase);
    case ARGREP_PRED_RE:
        return tre_regnexecb(&pred->regex, data, len, 0, NULL, 0) == REG_OK;
    default:
        serverPanic("Unknown ARGREP predicate type");
    }
}

/* Decode one array value and apply all the predicates to it. */
int arGrepValueMatches(arGrepPlan *plan, void *v) {
    char buf[AR_INLINE_BUFSIZE];
    size_t len;
    const char *data = arDecode(v, buf, sizeof(buf), &len);

    if (plan->combine == ARGREP_COMBINE_AND) {
        for (int i = 0; i < plan->num_preds; i++) {
            if (!arGrepMatchPredicate(&plan->preds[i], data, len,
                                      plan->nocase)) {
                return 0;
            }
        }
        return 1;
    }

    for (int i = 0; i < plan->num_preds; i++) {
        if (arGrepMatchPredicate(&plan->preds[i], data, len, plan->nocase))
            return 1;
    }
    return 0;
}

/* ARGREP key start end
 *        (EXACT string | MATCH string | GLOB pattern | RE pattern) ...
 *        [AND | OR] [LIMIT count] [WITHVALUES] [NOCASE]
 *
 * Search existing elements in a range and return matching indexes.
 *
 * Complexity is O(P * C), where P is the number of visited positions in the
 * touched slices and C is the cost of evaluating the active predicates.
 * Dense slices scan the touched dense window, sparse slices only visit stored
 * entries, and LIMIT stops as soon as enough matches were emitted.
 *
 * "-" and "+" mean the logical start and end of the array. WITHVALUES changes
 * the reply from [idx ...] to [idx value ...]. */
void argrepCommand(client *c) {
    arGrepBound start_bound, end_bound;
    if (arGrepParseBoundOrReply(c, c->argv[2], &start_bound) != C_OK) return;
    if (arGrepParseBoundOrReply(c, c->argv[3], &end_bound) != C_OK) return;

    arGrepPlan plan;
    uint64_t remaining;
    if (arGrepParsePlanOrReply(c, &plan, &remaining) != C_OK) {
        arGrepFreePlan(&plan);
        return;
    }

    robj *o = lookupKeyRead(c->db, c->argv[1]);
    if (o != NULL && checkType(c, o, OBJ_ARRAY)) {
        arGrepFreePlan(&plan);
        return;
    }
    if (o == NULL) {
        arGrepFreePlan(&plan);
        addReplyArrayLen(c, 0);
        return;
    }

    redisArray *ar = o->ptr;
    uint64_t ar_len = arLen(ar);
    if (ar_len == 0 || arCount(ar) == 0) {
        arGrepFreePlan(&plan);
        addReplyArrayLen(c, 0);
        return;
    }

    void *replylen = addReplyDeferredLen(c);
    uint64_t count = 0;
    uint64_t max_index = ar_len - 1;
    uint64_t start = arGrepResolveBound(&start_bound, max_index);
    uint64_t end = arGrepResolveBound(&end_bound, max_index);
    arScanIter it;
    uint64_t idx;
    void *v;

    arScanIterInit(ar, start, end, &it);
    while (remaining && arScanIterNext(&it, &idx, &v)) {
        if (!arGrepValueMatches(&plan, v)) continue;
        /* With WITHVALUES, reply nested [idx, value] pairs. */
        if (plan.withvalues) addReplyArrayLen(c, 2);
        addReplyUnsignedLongLong(c, idx);
        if (plan.withvalues) addReplyArrayValue(c, v);
        count++;
        remaining--;
    }

    setDeferredArrayLen(c, replylen, count);
    arGrepFreePlan(&plan);
}

/* ============================================================================
 * AROP
 * ============================================================================
 *
 * Aggregate operations over a range. Uses O(N) iteration where N is the
 * number of stored elements. Dense slices scan the window intersection
 * (bounded by dense.winsize, kept small by demotion when density drops).
 * -------------------------------------------------------------------------- */

/* Operation types for AROP */
#define AROP_SUM   1  /* Sum of numeric elements in range. */
#define AROP_MIN   2  /* Minimum numeric element in range. */
#define AROP_MAX   3  /* Maximum numeric element in range. */
#define AROP_AND   4  /* Bitwise AND of integer elements in range. */
#define AROP_OR    5  /* Bitwise OR of integer elements in range. */
#define AROP_XOR   6  /* Bitwise XOR of integer elements in range. */
#define AROP_MATCH 7  /* Count elements equal to a target string. */
#define AROP_USED  8  /* Count of non-empty (used) slots in range. */

/* Accumulator state for AROP */
typedef struct {
    int op;                    /* Selected AROP operation. */
    sds match_val;             /* MATCH target string. */
    long double sum_acc;       /* Running SUM accumulator. */
    long double minmax_acc;    /* Running MIN or MAX accumulator. */
    int64_t bitwise_acc;       /* Running AND/OR/XOR accumulator. */
    long long match_count;     /* Number of MATCH hits. */
    long long used_count;      /* Number of non-empty elements seen. */
    int has_numeric;           /* Saw at least one numeric value. */
    int has_int;               /* Saw at least one bitwise-usable integer. */
} arOpAcc;

/* Process a single value for AROP aggregation, aggregating it
 * into the structure arOpAcc 'acc'. This helper is used
 * directly by the AROP command implementation while scanning
 * populated elements in the requested range. */
static inline void arOpAccumulate(arOpAcc *acc, void *v) {
    if (acc->op == AROP_USED) {
        acc->used_count++;
        return;
    }

    if (acc->op == AROP_MATCH) {
        size_t vlen;
        char vbuf[AR_INLINE_BUFSIZE];
        const char *data = arDecode(v, vbuf, sizeof(vbuf), &vlen);
        if (vlen == sdslen(acc->match_val) &&
            memcmp(data, acc->match_val, vlen) == 0) {
            acc->match_count++;
        }
        return;
    }

    /* Numeric operations */
    long double num;
    int is_int = 0;
    int64_t ival = 0;

    if (arIsInt(v)) {
        ival = arToInt(v);
        num = (long double)ival;
        is_int = 1;
    } else if (arIsFloat(v)) {
        num = (long double)arToDouble(v);
    } else {
        const char *data;
        size_t vlen;
        char smallbuf[8];

        if (arIsSmallStr(v)) {
            vlen = arToSmallStr(v, smallbuf);
            data = smallbuf;
        } else {
            data = arStringData(v);
            vlen = arStringLen(v);
        }

        long long ll;
        if (string2ll(data, vlen, &ll)) {
            ival = ll;
            num = (long double)ll;
            is_int = 1;
        } else {
            long double ld;
            if (string2ld(data, vlen, &ld)) {
                num = ld;
            } else {
                return;
            }
        }
    }

    if (acc->op == AROP_AND || acc->op == AROP_OR || acc->op == AROP_XOR) {
        if (!is_int) {
            /* If it is a float, we only take the integer part. */
            if (isnan(num)) return;
            if (num < (long double)INT64_MIN || num > (long double)INT64_MAX)
                return;
            ival = (int64_t)num;  /* Truncate toward zero. */
        }
        if (!acc->has_int) {
            acc->bitwise_acc = ival;
            acc->has_int = 1;
        } else {
            if (acc->op == AROP_AND) acc->bitwise_acc &= ival;
            else if (acc->op == AROP_OR) acc->bitwise_acc |= ival;
            else acc->bitwise_acc ^= ival;
        }
    } else {
        if (!acc->has_numeric) {
            /* Handle the first element seen for SUM, MIN, MAX. */
            acc->sum_acc = num;
            acc->minmax_acc = num;
            acc->has_numeric = 1;
        } else {
            if (acc->op == AROP_SUM)
                acc->sum_acc += num;
            else if (acc->op == AROP_MIN && num < acc->minmax_acc)
                acc->minmax_acc = num;
            else if (acc->op == AROP_MAX && num > acc->minmax_acc)
                acc->minmax_acc = num;
        }
    }
}

/* AROP key start end OP [arg]
 *
 * Aggregates over existing elements in the requested range, the
 * aggregation performed depends in the "op" argument.
 *
 * Complexity is O(P), where P is visited positions in touched slices
 * (dense scanned slots + sparse entries), with worst-case O(|end-start|+1)
 * and typical case close to O(N), where N is the number of existing
 * elements in range.
 *
 * MATCH and USED count hits. SUM/MIN/MAX ignore values that are not numeric.
 * AND/OR/XOR truncate floats toward zero and ignore values that, after the
 * truncation, cannot be represented as int64_t. */
void aropCommand(client *c) {
    uint64_t start, end;
    if (arrayParseIndexOrReply(c, c->argv[2], &start) != C_OK) return;
    if (arrayParseIndexOrReply(c, c->argv[3], &end) != C_OK) return;

    const char *opstr = c->argv[4]->ptr;
    int op = 0;
    if (!strcasecmp(opstr, "SUM")) op = AROP_SUM;
    else if (!strcasecmp(opstr, "MIN")) op = AROP_MIN;
    else if (!strcasecmp(opstr, "MAX")) op = AROP_MAX;
    else if (!strcasecmp(opstr, "AND")) op = AROP_AND;
    else if (!strcasecmp(opstr, "OR")) op = AROP_OR;
    else if (!strcasecmp(opstr, "XOR")) op = AROP_XOR;
    else if (!strcasecmp(opstr, "MATCH")) op = AROP_MATCH;
    else if (!strcasecmp(opstr, "USED")) op = AROP_USED;
    else {
        addReplyError(c, "unknown operation");
        return;
    }

    sds match_val = NULL;
    if (op == AROP_MATCH) {
        if (c->argc != 6) {
            addReplyError(c, "MATCH requires a value argument");
            return;
        }
        match_val = c->argv[5]->ptr;
    } else if (c->argc != 5) {
        addReplyErrorArity(c);
        return;
    }

    robj *o = lookupKeyRead(c->db, c->argv[1]);
    if (o == NULL) {
        if (op == AROP_MATCH || op == AROP_USED) {
            addReplyLongLong(c, 0);
        } else {
            addReplyNull(c);
        }
        return;
    }
    if (checkType(c, o, OBJ_ARRAY)) return;

    redisArray *ar = o->ptr;
    arOpAcc acc = {
        .op = op, .match_val = match_val,
        .sum_acc = 0, .minmax_acc = 0, .bitwise_acc = 0,
        .match_count = 0, .used_count = 0,
        .has_numeric = 0, .has_int = 0
    };
    arScanIter it;
    void *v;

    /* All current AROP operations are order-independent, so iterating the
     * user-provided direction is fine here. */
    arScanIterInit(ar, start, end, &it);
    while (arScanIterNext(&it, NULL, &v))
        arOpAccumulate(&acc, v);

    /* Reply */
    if (op == AROP_MATCH) {
        addReplyLongLong(c, acc.match_count);
    } else if (op == AROP_USED) {
        addReplyLongLong(c, acc.used_count);
    } else if (op == AROP_AND || op == AROP_OR || op == AROP_XOR) {
        if (!acc.has_int) addReplyNull(c);
        else addReplyLongLong(c, acc.bitwise_acc);
    } else {
        if (!acc.has_numeric) {
            addReplyNull(c);
        } else {
            long double result = (op == AROP_SUM) ? acc.sum_acc : acc.minmax_acc;
            char buf[MAX_LONG_DOUBLE_CHARS + 1];
            int len = ld2string(buf, sizeof(buf), result, LD_STR_AUTO);
            addReplyBulkCBuffer(c, buf, len);
        }
    }
}

/* ----------------------------------------------------------------------------
 * The ring buffer family of commands:
 *
 * ARINSERT / ARNEXT / ARSEEK / ARLASTITEMS
 * -------------------------------------------------------------------------- */

/* ARINSERT key value [value ...]
 *
 * Appends one or more values at the private insert cursor in O(N), where N is
 * the number of values. The whole batch fails on index overflow.
 *
 * The cursor is then advanced to the last written index, which is also
 * returned as the command return value, and can be inspected later
 * with ARNEXT. */
void arinsertCommand(client *c) {
    robj *o = lookupArrayForWriteOrReply(c, c->argv[1]);
    if (o == NULL) return;

    redisArray *ar = o->ptr;
    uint64_t old_count = arCount(ar);
    size_t old_alloc = 0;
    if (server.memory_tracking_enabled) old_alloc = kvobjAllocSize(o);
    int num_values = c->argc - 2;

    /* Pre-validate: compute start cursor and check entire batch fits */
    uint64_t start_cursor;
    if (ar->insert_idx == AR_INSERT_IDX_NONE) {
        start_cursor = 0;
    } else {
        if (ar->insert_idx >= UINT64_MAX - 1) {
            addReplyError(c, "insert index overflow");
            return;
        }
        start_cursor = ar->insert_idx + 1;
    }

    /* Check last cursor won't overflow or reach forbidden index. */
    uint64_t last_cursor = start_cursor + (uint64_t)num_values - 1;
    if (last_cursor < start_cursor || last_cursor == UINT64_MAX) {
        addReplyError(c, "insert index overflow");
        return;
    }

    /* Pre-promote sparse slices only for true bulk inserts. A single-element
     * insert does not benefit from the extra range-analysis pass. */
    if (num_values > 1)
        arMayPromoteToDenseForRangeSet(ar, start_cursor, last_cursor);

    /* Apply all values */
    uint64_t cursor = start_cursor;
    for (int i = 0; i < num_values; i++) {
        sds val = c->argv[2 + i]->ptr;
        void *v = arEncode(val, sdslen(val));
        arSet(ar, cursor, v);
        cursor++;
    }
    ar->insert_idx = last_cursor;

    updateKeysizesHist(c->db, OBJ_ARRAY, old_count, arCount(ar));
    if (server.memory_tracking_enabled)
        updateSlotAllocSize(c->db, getKeySlot(c->argv[1]->ptr), o, old_alloc, kvobjAllocSize(o));
    keyModified(c, c->db, c->argv[1], o, 1);
    notifyKeyspaceEvent(NOTIFY_ARRAY, "arinsert", c->argv[1], c->db->id);
    server.dirty += num_values;

    addReplyUnsignedLongLong(c, ar->insert_idx);
}

/* Duplicate one array value exactly. Immediate values can be copied as tagged
 * words, while heap strings are re-encoded from their logical string form.
 * This could be regarded as costly, but capturing values out of the existing
 * array would break the sparsearray API isolation. */
static void *arRingDupValue(void *v) {
    if (v == NULL || !arIsPtr(v)) return v;
    return arEncode(arStringData(v), arStringLen(v));
}

/* Return the next slot that ARRING would write to before modulo reduction. */
static uint64_t arRingNextCursor(redisArray *ar) {
    return (ar->insert_idx == AR_INSERT_IDX_NONE) ? 0 : ar->insert_idx + 1;
}

/* Decide if ARRING needs to rebuild the retained logical ring positions before
 * writing new values.
 *
 * We rebuild in only two cases:
 *
 * 1. Shrink: new size is smaller than the current inferred ring span.
 * 2. Grow after wrap: the ring had already wrapped inside the old span, so
 *    without a rebuild the next write would overwrite old low indexes instead
 *    of using the newly added capacity.
 *
 * An explicit ARSEEK 0 is treated differently on grow: it is a direct cursor
 * override saying "write next at index 0", so we honor it instead of forcing
 * a grow-after-wrap repack first.
 *
 * keep_span is the maximum number of logical positions that may be retained. */
static int arRingNeedsRework(redisArray *ar, uint64_t ring_size,
                             uint64_t *old_span, uint64_t *keep_span) {
    *old_span = arLen(ar);
    *keep_span = 0;

    if (*old_span == 0) return 0;

    if (ring_size < *old_span) {
        *keep_span = ring_size;
        return 1;
    }
    if (ring_size == *old_span) {
        return 0;
    }
    if (ar->insert_idx == AR_INSERT_IDX_NONE) {
        return 0;
    }
    if (arRingNextCursor(ar) < *old_span) {
        *keep_span = *old_span;
        return 1;
    }
    return 0;
}

/* Rebuild the retained logical ring positions into a fresh compact array.
 *
 * We walk backward from the current anchor and keep at most keep_span items,
 * but stop as soon as the first NULL is encountered. This makes resize keep
 * the latest contiguous tail of existing items instead of crossing holes.
 *
 * The retained items are replayed in chronological order, oldest to newest,
 * so after the rebuild:
 *
 * - index 0 holds the oldest retained position
 * - index retained_count-1 holds the newest retained position
 * - insert_idx points to retained_count-1, ready for the next ARRING write
 *
 * We use two passes: one backward pass to count the contiguous retained tail,
 * then one forward replay pass into the new array. This avoids any temporary
 * retained-items buffer. */
static redisArray *arRingRework(redisArray *ar, uint64_t old_span,
                                uint64_t keep_span) {
    serverAssert(old_span > 0);
    serverAssert(keep_span > 0);
    serverAssert(keep_span <= old_span);

    redisArray *new_ar = arNew();

    /* The rebuild operates on the inferred ring window [0..old_span-1]. If
     * insert_idx is outside that window because of ARSEEK, fold it back into
     * the current inferred span with modulo. If ARSEEK 0 was used and we are
     * shrinking, anchor the walk at the current tail, just like ARLASTITEMS.
     * Grow does not reach this path because arRingNeedsRework() skips grow
     * rework when insert_idx is AR_INSERT_IDX_NONE. */
    uint64_t anchor_idx = (ar->insert_idx == AR_INSERT_IDX_NONE) ?
        (old_span - 1) : (ar->insert_idx % old_span);

    uint64_t retained_count = 0;
    uint64_t src_idx = anchor_idx;

    while (retained_count < keep_span) {
        void *v = arGet(ar, src_idx);
        if (v == NULL) break; /* This makes any mix of ARSET/SEEK/RING calls
                               * always bound to populatede items, not logical
                               * array span. */

        retained_count++;
        src_idx = (src_idx == 0) ? old_span - 1 : src_idx - 1;
    }

    /* src_idx now points to the position just before the oldest retained
     * item, so advance once to start replaying oldest -> newest. */
    src_idx++;
    if (src_idx == old_span) src_idx = 0;

    for (uint64_t dst_idx = 0; dst_idx < retained_count; dst_idx++) {
        void *v = arGet(ar, src_idx);
        serverAssert(v != NULL);
        arSet(new_ar, dst_idx, arRingDupValue(v));

        src_idx++;
        if (src_idx == old_span) src_idx = 0;
    }
    if (retained_count != 0) new_ar->insert_idx = retained_count - 1;
    return new_ar;
}

/* ARRING key size value [value ...]
 *
 * Writes values into a logical ring buffer. May rework the array if
 * the logical size changes across calls, so that the up to size
 * items are retained in the correct logical position.
 *
 * Complexity is O(M) normally, where M is the number of inserted values,
 * and O(N+M) on resize, where N is the maximum of the old and new ring size.
 * The rebuild stops at the first NULL, so holes cut the retained tail.
 *
 * ARSEEK 0 is still honored as a direct cursor override on grow.
 *
 * Returns the last written slot. */
void arringCommand(client *c) {
    long long ll;
    if (getLongLongFromObjectOrReply(c,c->argv[2],&ll,"invalid size") != C_OK)
        return;
    if (ll <= 0) {
        addReplyError(c, "size must be positive");
        return;
    }
    uint64_t ring_size = (uint64_t)ll;

    robj *o = lookupArrayForWriteOrReply(c, c->argv[1]);
    if (o == NULL) return;

    redisArray *ar = o->ptr;
    uint64_t old_count = arCount(ar);
    size_t old_alloc = 0;
    if (server.memory_tracking_enabled) old_alloc = kvobjAllocSize(o);
    int num_values = c->argc - 3;
    uint64_t cursor = 0;

    /* If the requested size changes the logical ring shape, rebuild once
     * before the hot insertion loop. This makes the command, when the user
     * updates the window, no longer O(M), but O(N+M), however note that this
     * is absolutely needed for high level sane semantics. Users will resize
     * ring buffers, and they want to retain the latest items in a logically
     * correct way. */
    uint64_t old_span, keep_span;
    if (arRingNeedsRework(ar, ring_size, &old_span, &keep_span)) {
        redisArray *new_ar = arRingRework(ar, old_span, keep_span);
        arFree(ar);
        o->ptr = ar = new_ar;
    }

    /* Set the new items, modulo ring size. */
    for (int i = 0; i < num_values; i++) {
        /* Compute the next write position, then wrap it into the requested
         * ring size if needed. By this point any needed resize/rework was
         * already handled above. */
        cursor = arRingNextCursor(ar);
        if (cursor >= ring_size) cursor = cursor % ring_size;

        /* Set the value */
        sds val = c->argv[3 + i]->ptr;
        void *v = arEncode(val, sdslen(val));
        arSet(ar, cursor, v);
        ar->insert_idx = cursor;
    }

    updateKeysizesHist(c->db, OBJ_ARRAY, old_count, arCount(ar));
    if (server.memory_tracking_enabled)
        updateSlotAllocSize(c->db, getKeySlot(c->argv[1]->ptr), o, old_alloc, kvobjAllocSize(o));
    keyModified(c, c->db, c->argv[1], o, 1);
    notifyKeyspaceEvent(NOTIFY_ARRAY, "arring", c->argv[1], c->db->id);
    server.dirty += num_values;

    addReplyUnsignedLongLong(c, cursor);
}

/* ARNEXT key
 *
 * Returns in O(1) the next index that ARINSERT / ARRING would use.
 *
 * Missing keys and the pre-insert state reply with 0. If the cursor is in the
 * terminal state where the next append would overflow, the reply is NULL. */
void arnextCommand(client *c) {
    robj *o = lookupKeyRead(c->db, c->argv[1]);
    if (o == NULL) {
        addReplyLongLong(c, 0);
        return;
    }
    if (checkType(c, o, OBJ_ARRAY)) return;

    redisArray *ar = o->ptr;
    if (ar->insert_idx == AR_INSERT_IDX_NONE) {
        addReplyLongLong(c, 0);
    } else if (ar->insert_idx == UINT64_MAX - 1) {
        addReplyNull(c);  /* Terminal: index space exhausted */
    } else {
        addReplyUnsignedLongLong(c, ar->insert_idx + 1);
    }
}

/* ARSEEK key idx
 *
 * Sets in O(1) the next index used by ARINSERT and ARRING.
 *
 * Returns 1 if the cursor was updated and 0 if the key does not exist.
 * idx 0 resets the insert state to "next write goes to 0": in this case
 * successive ARRING calls are guaranteed to don't rework the array in chase
 * of logical size change. */
void arseekCommand(client *c) {
    uint64_t idx;
    /* Allow UINT64_MAX because ARSEEK UINT64_MAX sets insert_idx to
     * UINT64_MAX-1, which is a valid terminal state (next ARINSERT
     * would overflow and fail). This is needed for AOF persistence. */
    if (getArrayIndexFromObject(c->argv[2], &idx, 1) != C_OK) {
        addReplyError(c, "invalid array index");
        return;
    }

    /* There aren't many good options for non existing keys: both creating
     * an empty array or failing with "no such key" does not align very
     * well with the Redis commands usual semantics. However we need to signal
     * back that we ignored the index set if the key is not there, so zero
     * is returned. */
    robj *o = lookupKeyWrite(c->db, c->argv[1]);
    if (o == NULL) {
        addReplyLongLong(c, 0);
        return;
    }
    if (checkType(c, o, OBJ_ARRAY)) return;

    redisArray *ar = o->ptr;

    /* Set insert_idx so next ARINSERT writes to idx */
    if (idx == 0) {
        ar->insert_idx = AR_INSERT_IDX_NONE;
    } else {
        ar->insert_idx = idx - 1;
    }

    keyModified(c, c->db, c->argv[1], o, 1);
    notifyKeyspaceEvent(NOTIFY_ARRAY, "arseek", c->argv[1], c->db->id);
    server.dirty++;
    addReplyLongLong(c, 1);
}

/* ARLASTITEMS key count [REV]
 *
 * Returns the most recent positions from the current insert anchor in O(N),
 * where N is the requested count. REV flips the reply order.
 *
 * This command may return NULLs because it walks positions, not only existing
 * items. If ARSEEK 0 was used, the current array tail is used as the anchor. */
void arlastitemsCommand(client *c) {
    long long count;
    if (getLongLongFromObjectOrReply(c, c->argv[2], &count,
        "invalid COUNT") != C_OK) return;

    /* For count <= 0, nothing to return, just an empty array. */
    if (count <= 0) {
        addReplyArrayLen(c, 0);
        return;
    }

    /* Parse REV if provided. */
    int rev = 0;
    if (c->argc == 4) {
        if (strcasecmp(c->argv[3]->ptr, "REV") == 0) {
            rev = 1;
        } else {
            addReplyErrorObject(c, shared.syntaxerr);
            return;
        }
    } else if (c->argc != 3) {
        addReplyErrorArity(c);
        return;
    }

    /* No key? Empty reply. */
    robj *o = lookupKeyRead(c->db, c->argv[1]);
    if (o == NULL) {
        addReplyArrayLen(c, 0);
        return;
    }
    if (checkType(c, o, OBJ_ARRAY)) return;

    redisArray *ar = o->ptr;
    uint64_t ar_len = arLen(ar);
    uint64_t effective_count =
        (uint64_t)count > ar->count ? ar->count : (uint64_t)count;

    /* Should never happen in practice, because we checked the COUNT before
     * and the array should not be empty to be still a Redis key, so this
     * is mostly a safety net. */
    if (effective_count == 0) {
        addReplyArrayLen(c, 0);
        return;
    }

    /* Collect items walking backward from insert_idx. If ARSEEK 0 was used,
     * insert_idx is AR_INSERT_IDX_NONE: in that case use the max set index as
     * the anchor so ARLASTITEMS still reports the tail of the current array.
     *
     * Note that we use an array to collect the items: in the no-REV case
     * otherwise a double scan would be needed. */
    void **collected = zmalloc(effective_count * sizeof(void *));
    uint64_t anchor_idx =
        (ar->insert_idx == AR_INSERT_IDX_NONE) ? ar_len - 1 : ar->insert_idx;
    uint64_t current_idx = anchor_idx;
    uint64_t steps = 0;

    while(steps < effective_count) {
        collected[steps] = arGet(ar, current_idx);
        steps++;

        /* Decrement with wrap */
        if (current_idx == 0) {
            current_idx = ar_len - 1;
        } else {
            current_idx--;
        }
    }

    /* Emit the protocol with the collected items. */
    addReplyArrayLen(c, steps);
    if (rev) {
        /* Return in reverse chronological order (newest first) */
        for (uint64_t i = 0; i < steps; i++)
            addReplyArrayValue(c, collected[i]);
    } else {
        /* Return in chronological order (oldest first) */
        for (int64_t i = steps - 1; i >= 0; i--)
            addReplyArrayValue(c, collected[i]);
    }
    zfree(collected);
}

/* ----------------------------------------------------------------------------
 * ARINFO
 * -------------------------------------------------------------------------- */

/* ARINFO key [FULL]
 *
 * Returns metadata about the array in O(1), or O(N) with FULL where N is the
 * number of slices. Unlike ARLEN and ARCOUNT, a missing key is an error.
 * FULL adds per-encoding slice statistics by scanning the directory. */
void arinfoCommand(client *c) {
    int full = 0;

    if (c->argc > 2) {
        if (c->argc == 3 && !strcasecmp(c->argv[2]->ptr, "full")) {
            full = 1;
        } else {
            addReplyErrorObject(c, shared.syntaxerr);
            return;
        }
    }

    robj *o = lookupKeyRead(c->db, c->argv[1]);
    if (o == NULL) {
        addReplyError(c, "no such key");
        return;
    }
    if (checkType(c, o, OBJ_ARRAY)) return;

    redisArray *ar = o->ptr;

    /* Per-encoding stats (only computed for FULL) */
    uint64_t num_dense = 0;
    uint64_t num_sparse = 0;
    uint64_t dense_total_winsize = 0;
    uint64_t dense_total_count = 0;
    uint64_t sparse_total_cap = 0;

    if (full) {
        if (ar->superdir) {
            for (uint32_t bi = 0; bi < ar->sdir_len; bi++) {
                arSDirEntry *e = ar->superdir + bi;
                for (uint32_t si = 0; si < AR_SUPER_BLOCK_SLOTS; si++) {
                    arSlice *s = e->slots[si];
                    if (!s) continue;
                    if (s->encoding == AR_SLICE_DENSE) {
                        num_dense++;
                        dense_total_winsize += s->layout.dense.winsize;
                        dense_total_count += s->count;
                    } else {
                        num_sparse++;
                        sparse_total_cap += s->layout.sparse.cap;
                    }
                }
            }
        } else {
            for (uint64_t i = 0; i < ar->dir_alloc; i++) {
                arSlice *s = ar->dir[i];
                if (!s) continue;
                if (s->encoding == AR_SLICE_DENSE) {
                    num_dense++;
                    dense_total_winsize += s->layout.dense.winsize;
                    dense_total_count += s->count;
                } else {
                    num_sparse++;
                    sparse_total_cap += s->layout.sparse.cap;
                }
            }
        }
    }

    if (full) {
        addReplyMapLen(c, 12);
    } else {
        addReplyMapLen(c, 7);
    }

    addReplyBulkCString(c, "count");
    addReplyUnsignedLongLong(c, ar->count);

    addReplyBulkCString(c, "len");
    addReplyUnsignedLongLong(c, arLen(ar));

    addReplyBulkCString(c, "next-insert-index");
    if (ar->insert_idx == AR_INSERT_IDX_NONE ||
        ar->insert_idx == UINT64_MAX - 1) {
        addReplyLongLong(c, 0);
    } else {
        addReplyUnsignedLongLong(c, ar->insert_idx + 1);
    }

    addReplyBulkCString(c, "slices");
    addReplyLongLong(c, ar->num_slices);

    addReplyBulkCString(c, "directory-size");
    if (ar->superdir) {
        /* Superdir mode: report allocated capacity */
        addReplyLongLong(c, ar->sdir_cap);
    } else {
        addReplyLongLong(c, ar->dir_alloc);
    }

    addReplyBulkCString(c, "super-dir-entries");
    addReplyLongLong(c, ar->superdir ? ar->sdir_len : 0);

    addReplyBulkCString(c, "slice-size");
    addReplyLongLong(c, ar->slice_size);

    if (full) {
        addReplyBulkCString(c, "dense-slices");
        addReplyLongLong(c, num_dense);

        addReplyBulkCString(c, "sparse-slices");
        addReplyLongLong(c, num_sparse);

        addReplyBulkCString(c, "avg-dense-size");
        if (num_dense > 0) {
            addReplyDouble(c, (double)dense_total_winsize / num_dense);
        } else {
            addReplyDouble(c, 0);
        }

        addReplyBulkCString(c, "avg-dense-fill");
        if (dense_total_winsize > 0) {
            addReplyDouble(c, (double)dense_total_count / dense_total_winsize);
        } else {
            addReplyDouble(c, 0);
        }

        addReplyBulkCString(c, "avg-sparse-size");
        if (num_sparse > 0) {
            addReplyDouble(c, (double)sparse_total_cap / num_sparse);
        } else {
            addReplyDouble(c, 0);
        }
    }
}