Wrong inheritance of the delete API with async Redis cluster pipeline

[karwinliu]

Version: v4.3.4

Platform: Python 3.9 on linux/macOS

Description: When delete() is called in a pipeline (async cluster version), it unexpectedly hits the delete() implementation of ClusterMultiKeyCommands class below,

    def delete(self, *keys: KeyT) -> ResponseT:
        """
        Deletes the given keys in the cluster.
        The keys are first split up into slots
        and then an DEL command is sent for every slot

        Non-existant keys are ignored.
        Returns the number of keys that were deleted.

        For more information see https://redis.io/commands/del
        """
        return self._split_command_across_slots("DEL", *keys)

This is how it is used on our end,

    async with c.pipeline() as p:
        for key in keys:
            p.delete(key)
        return await p.execute()

The reason we use pipeline instead of the multi-key delete is because keys are split by slots under the hood, and it is less performance than splitting by node that is already handled by the pipeline. The above behavior should be similar to what's being done in sync pipeline,

    def delete(self, *names):
        """
        "Delete a key specified by ``names``"
        """
        if len(names) != 1:
            raise RedisClusterException(
                "deleting multiple keys is not " "implemented in pipeline command"
            )

        return self.execute_command("DEL", names[0])

[utkarshgupta137]

I don't think you can include keys from multiple slots in the same delete command, but ultimately all the commands are sent over the same connection. I'll check if keys from multiples slots can be included in the same command.

[karwinliu]

I thought currently if I call async RedisCLuster's delete() with multiple keys, it would still work -- first split keys by slot and then send aggregated commands to each slot. The problem is if multiple slots belong to the same node, are they sent in parallel in a single round trip? In another word, will it be at most 16384 (slot count) or N round trips (N is the node count)?

If yes, then I can just directly use that, but if not, I'd hope to work around by wrapping all the delete commands in a pipeline, because I know the pipeline would aggregate by node, so there will only be at most N round trips

[karwinliu]

After taking a closer look at the code, the async RedisCluster's delete(*keys) would eventually aggregate over node and send everything in a pipeline, which is essentially very similar to the following,

async with c.pipeline() as p:
    for key in keys:
        p.delete(key)
    return await p.execute()

So I can just use the delete API directly.
However, it is a common use case to have delete and other different commands in a pipeline, and it should simply trigger something like below,

def delete(self, name):
    return self.execute_command("DEL", name)

instead of currently

def delete(self, *keys: KeyT) -> ResponseT:
    """
    Deletes the given keys in the cluster.
    The keys are first split up into slots
    and then an DEL command is sent for every slot

    Non-existant keys are ignored.
    Returns the number of keys that were deleted.

    For more information see https://redis.io/commands/del
    """
    return self._split_command_across_slots("DEL", *keys)

which will throw errors.
Let me know if it makes sense, it should be pretty easy to repro, just need to run the following,

async with c.pipeline() as p:
    p.get(key)
    p.delete(key)
return await p.execute()

[utkarshgupta137]

I thought currently if I call async RedisCLuster's delete() with multiple keys, it would still work -- first split keys by slot and then send aggregated commands to each slot. The problem is if multiple slots belong to the same node, are they sent in parallel in a single round trip? In another word, will it be at most 16384 (slot count) or N round trips (N is the node count)?

If yes, then I can just directly use that, but if not, I'd hope to work around by wrapping all the delete commands in a pipeline, because I know the pipeline would aggregate by node, so there will only be at most N round trips

I can confirm that you will get a CROSSSLOT error if keys from multiple slots are in the same command. You don't want to send parallel requests to each node for each slot. Otherwise, what's the point of the pipeline. You want to send multiple commands to the same node over a single connection in a single call, which improves performance dramatically.

[utkarshgupta137]

After taking a closer look at the code, the async RedisCluster's delete(*keys) would eventually aggregate over node and send everything in a pipeline, which is essentially very similar to the following,

async with c.pipeline() as p:
    for key in keys:
        p.delete(key)
    return await p.execute()

So I can just use the delete API directly. However, it is a common use case to have delete and other different commands in a pipeline, and it should simply trigger something like below,

def delete(self, name):
    return self.execute_command("DEL", name)

instead of currently

def delete(self, *keys: KeyT) -> ResponseT:
    """
    Deletes the given keys in the cluster.
    The keys are first split up into slots
    and then an DEL command is sent for every slot

    Non-existant keys are ignored.
    Returns the number of keys that were deleted.

    For more information see https://redis.io/commands/del
    """
    return self._split_command_across_slots("DEL", *keys)

which will throw errors. Let me know if it makes sense, it should be pretty easy to repro, just need to run the following,

async with c.pipeline() as p:
    p.get(key)
    p.delete(key)
return await p.execute()

I am not sure what you're asking for. If you are talking about having both get, set, and delete operations in the same pipeline, then the behaviour would be the same. You can refer to the below screenshot for the order that you will get.

[karwinliu]

@utkarshgupta137 I did a double check and it seems to be working now both with async RedisCluster (e.g. delete(*keys)) and async ClusterPipeline(e.g. p.delete(*keys)) after apply the fix #2217. Something must be in the mix that led to the wrong interpretation of the delete command. Thanks for helping anyways and I will close this one