feat(sink): support redis sink stream

[xxhZs]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

Checklist

• I have written necessary rustdoc comments.
• I have added necessary unit tests and integration tests.
• I have added test labels as necessary.
• I have added fuzzing tests or opened an issue to track them.
• My PR contains breaking changes.
• My PR changes performance-critical code, so I will run (micro) benchmarks and present the results.
• I have checked the Release Timeline and Currently Supported Versions to determine which release branches I need to cherry-pick this PR into.

Documentation

• My PR needs documentation updates.

Release note

We can use pubsub like

create table t1(v1 float, v2 float,v3 varchar, v4 varchar);
CREATE SINK s1
FROM
    t1 WITH (
    primary_key = 'v3,v4',
    connector = 'redis',
    redis.url= 'redis://127.0.0.1:6379/',
)FORMAT plain ENCODE TEMPLATE
(redis_value_type ='stream', 
stream = 'test123',
key_format = abcd3:{v3}',
value_format = 'abcd3:{v4}'
);

or

CREATE SINK s1
FROM
    t1 WITH (
    primary_key = 'v2,v3,v4',
    connector = 'redis',
    redis.url= 'redis://127.0.0.1:6379/',
)FORMAT plain ENCODE TEMPLATE
(redis_value_type ='stream', 
stream_column = 'v2',
key_format = abcd3:{v3}',
value_format = 'abcd3:{v4}'
);
insert into t1 values(1.1, 1.1, 'test','test');

value_format Required: Format of the value passed to stream
key_format Required: Format of the key passed to stream
stream stream_column both must set at least one.
stream The name of the message stream to send to
stream_column Use the values in this column as the name of the message stream to send to (type must be varchar)

[Copilot]

Pull Request Overview

This PR adds support for Redis Stream data type to the Redis sink connector in RisingWave, expanding beyond the existing support for strings, geospatial data, and pub/sub channels.

• Added REDIS_VALUE_TYPE_STREAM support for Redis Streams
• Implemented stream configuration with stream and stream_column parameters
• Added template encoders for Redis Stream key-value pairs

Reviewed Changes

Copilot reviewed 4 out of 5 changed files in this pull request and generated 1 comment.

File	Description
src/connector/src/sink/redis.rs	Added stream constants and pipeline operations for Redis streams
src/connector/src/sink/formatter/mod.rs	Added stream value type handling in template encoder build logic
src/connector/src/sink/encoder/template.rs	Implemented stream encoders and renamed PubSub encoders for consistency
src/connector/Cargo.toml	Added "streams" feature to redis dependency

Comments suppressed due to low confidence (1)

src/connector/src/sink/formatter/mod.rs:1

• Missing space after comma in error messages. Should be 'Cannot find '{VALUE_FORMAT}', please set it.' and 'Cannot find '{KEY_FORMAT}', please set it.'

// Copyright 2025 RisingWave Labs

[hzxa21]

nits:

Please specific either `{STREAM}` or `{STREAM_COLUMN}`. They are mutually exclusive options.

[hzxa21]

Is this expected? Should we add .vscode in gitignore?

[hzxa21]

LGTM

[hzxa21]

Please update the PR description. Also, please mention that this is a license feature in the PR description and the release note.

[tabVersion]

also request review from redis expert @shanicky :doge

[tabVersion]

No functionality tests on CI?

[hzxa21]

No functionality tests on CI?

@xxhZs can we add redis stream e2e test? It seems that this feature is available in open source edition.