Skip to content

Add RedisTimeSeries (TS.*) support #3852

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
HwangRock wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Add RedisTimeSeries (TS.*) support #3852

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
31e84b6
Add RedisTimeSeries (TS.*) support #2716
HwangRock Jul 18, 2026
16f2f04
Fix TS.* cluster integration tests #2716
HwangRock Jul 18, 2026
e745b62
Gate TS.* NaN edge-case tests on Redis 8.6 #2716
HwangRock Jul 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
150 changes: 149 additions & 1 deletion src/main/java/io/lettuce/core/AbstractRedisAsyncCommands.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,16 @@
import io.lettuce.core.search.arguments.SugAddArgs;
import io.lettuce.core.search.arguments.SugGetArgs;
import io.lettuce.core.search.arguments.SynUpdateArgs;
import io.lettuce.core.timeseries.TsAggregationType;
import io.lettuce.core.timeseries.TsInfoValue;
import io.lettuce.core.timeseries.TsMGetValue;
import io.lettuce.core.timeseries.TsSample;
import io.lettuce.core.timeseries.arguments.TsAddArgs;
import io.lettuce.core.timeseries.arguments.TsAlterArgs;
import io.lettuce.core.timeseries.arguments.TsCreateArgs;
import io.lettuce.core.timeseries.arguments.TsGetArgs;
import io.lettuce.core.timeseries.arguments.TsIncrByArgs;
import io.lettuce.core.timeseries.arguments.TsMGetArgs;
import io.lettuce.core.vector.RawVector;
import io.lettuce.core.vector.VSimScoreAttribs;
import io.lettuce.core.vector.VectorMetadata;
Expand Down Expand Up @@ -123,7 +133,8 @@ public abstract class AbstractRedisAsyncCommands<K, V> implements RedisAclAsyncC
RedisHLLAsyncCommands<K, V>, BaseRedisAsyncCommands<K, V>, RedisTransactionalAsyncCommands<K, V>,
RedisGeoAsyncCommands<K, V>, RedisClusterAsyncCommands<K, V>, RedisJsonAsyncCommands<K, V>,
RedisVectorSetAsyncCommands<K, V>, RediSearchAsyncCommands<K, V>, RedisArrayAsyncCommands<K, V>,
RedisBloomFilterAsyncCommands<K, V>, RedisCuckooFilterAsyncCommands<K, V>, RedisTopKAsyncCommands<K, V> {
RedisBloomFilterAsyncCommands<K, V>, RedisCuckooFilterAsyncCommands<K, V>, RedisTopKAsyncCommands<K, V>,
RedisTimeSeriesAsyncCommands<K, V> {

private final StatefulConnection<K, V> connection;

Expand All @@ -143,6 +154,8 @@ public abstract class AbstractRedisAsyncCommands<K, V> implements RedisAclAsyncC

private final RedisTopKCommandBuilder<K, V> topKCommandBuilder;

private final RedisTimeSeriesCommandBuilder<K, V> timeSeriesCommandBuilder;

private final Supplier<JsonParser> parser;

/**
Expand All @@ -164,6 +177,7 @@ public AbstractRedisAsyncCommands(StatefulConnection<K, V> connection, RedisCode
this.bloomFilterCommandBuilder = new RedisBloomFilterCommandBuilder<>(codec);
this.cuckooFilterCommandBuilder = new RedisCuckooFilterCommandBuilder<>(codec);
this.topKCommandBuilder = new RedisTopKCommandBuilder<>(codec);
this.timeSeriesCommandBuilder = new RedisTimeSeriesCommandBuilder<>(codec);
}

/**
Expand Down Expand Up @@ -4483,4 +4497,138 @@ public RedisFuture<String> topKReserve(K key, long k, TopKReserveArgs args) {
return dispatch(topKCommandBuilder.topKReserve(key, k, args));
}

// --- Redis Time Series Commands ---

@Override
public RedisFuture<String> tsCreate(K key) {
return dispatch(timeSeriesCommandBuilder.tsCreate(key));
}

@Override
public RedisFuture<String> tsCreate(K key, TsCreateArgs createArgs) {
return dispatch(timeSeriesCommandBuilder.tsCreate(key, createArgs));
}

@Override
public RedisFuture<String> tsAlter(K key, TsAlterArgs alterArgs) {
return dispatch(timeSeriesCommandBuilder.tsAlter(key, alterArgs));
}

@Override
public RedisFuture<String> tsCreateRule(K sourceKey, K destKey, TsAggregationType aggregationType, long bucketDuration) {
return dispatch(timeSeriesCommandBuilder.tsCreateRule(sourceKey, destKey, aggregationType, bucketDuration));
}

@Override
public RedisFuture<String> tsCreateRule(K sourceKey, K destKey, TsAggregationType aggregationType, long bucketDuration,
long alignTimestamp) {
return dispatch(
timeSeriesCommandBuilder.tsCreateRule(sourceKey, destKey, aggregationType, bucketDuration, alignTimestamp));
}

@Override
public RedisFuture<String> tsDeleteRule(K sourceKey, K destKey) {
return dispatch(timeSeriesCommandBuilder.tsDeleteRule(sourceKey, destKey));
}

@Override
public RedisFuture<Long> tsDel(K key, long fromTimestamp, long toTimestamp) {
return dispatch(timeSeriesCommandBuilder.tsDel(key, fromTimestamp, toTimestamp));
}

@Override
public RedisFuture<Long> tsAdd(K key, long timestamp, double value) {
return dispatch(timeSeriesCommandBuilder.tsAdd(key, timestamp, value));
}

@Override
public RedisFuture<Long> tsAdd(K key, long timestamp, double value, TsAddArgs addArgs) {
return dispatch(timeSeriesCommandBuilder.tsAdd(key, timestamp, value, addArgs));
}

@Override
public RedisFuture<Long> tsAdd(K key, double value) {
return dispatch(timeSeriesCommandBuilder.tsAdd(key, value));
}

@Override
public RedisFuture<List<Long>> tsMAdd(Map.Entry<K, TsSample>... entries) {
return dispatch(timeSeriesCommandBuilder.tsMAdd(entries));
}

@Override
public RedisFuture<List<Long>> tsMAdd(Map.Entry<K, TsSample> entry) {
return dispatch(timeSeriesCommandBuilder.tsMAdd(entry));
}

@Override
public RedisFuture<Long> tsIncrBy(K key, double value) {
return dispatch(timeSeriesCommandBuilder.tsIncrBy(key, value));
}

@Override
public RedisFuture<Long> tsIncrBy(K key, double value, TsIncrByArgs incrByArgs) {
return dispatch(timeSeriesCommandBuilder.tsIncrBy(key, value, incrByArgs));
}

@Override
public RedisFuture<Long> tsDecrBy(K key, double value) {
return dispatch(timeSeriesCommandBuilder.tsDecrBy(key, value));
}

@Override
public RedisFuture<Long> tsDecrBy(K key, double value, TsIncrByArgs decrByArgs) {
return dispatch(timeSeriesCommandBuilder.tsDecrBy(key, value, decrByArgs));
}

@Override
public RedisFuture<TsSample> tsGet(K key) {
return dispatch(timeSeriesCommandBuilder.tsGet(key));
}

@Override
public RedisFuture<TsSample> tsGet(K key, TsGetArgs getArgs) {
return dispatch(timeSeriesCommandBuilder.tsGet(key, getArgs));
}

@Override
public RedisFuture<List<TsMGetValue<K>>> tsMGet(V... filters) {
return dispatch(timeSeriesCommandBuilder.tsMGet(filters));
}

@Override
public RedisFuture<List<TsMGetValue<K>>> tsMGet(V filter) {
return dispatch(timeSeriesCommandBuilder.tsMGet(filter));
}

@Override
public RedisFuture<List<TsMGetValue<K>>> tsMGet(TsMGetArgs mGetArgs, V... filters) {
return dispatch(timeSeriesCommandBuilder.tsMGet(mGetArgs, filters));
}

@Override
public RedisFuture<List<TsMGetValue<K>>> tsMGet(TsMGetArgs mGetArgs, V filter) {
return dispatch(timeSeriesCommandBuilder.tsMGet(mGetArgs, filter));
}

@Override
public RedisFuture<TsInfoValue<K>> tsInfo(K key) {
return dispatch(timeSeriesCommandBuilder.tsInfo(key));
}

@Override
public RedisFuture<TsInfoValue<K>> tsInfoDebug(K key) {
return dispatch(timeSeriesCommandBuilder.tsInfoDebug(key));
}

@Override
public RedisFuture<List<K>> tsQueryIndex(V... filters) {
return dispatch(timeSeriesCommandBuilder.tsQueryIndex(filters));
}

@Override
public RedisFuture<List<K>> tsQueryIndex(V filter) {
return dispatch(timeSeriesCommandBuilder.tsQueryIndex(filter));
}

}
163 changes: 155 additions & 8 deletions src/main/java/io/lettuce/core/AbstractRedisReactiveCommands.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,16 @@
import io.lettuce.core.search.arguments.SugAddArgs;
import io.lettuce.core.search.arguments.SugGetArgs;
import io.lettuce.core.search.arguments.SynUpdateArgs;
import io.lettuce.core.timeseries.TsAggregationType;
import io.lettuce.core.timeseries.TsInfoValue;
import io.lettuce.core.timeseries.TsMGetValue;
import io.lettuce.core.timeseries.TsSample;
import io.lettuce.core.timeseries.arguments.TsAddArgs;
import io.lettuce.core.timeseries.arguments.TsAlterArgs;
import io.lettuce.core.timeseries.arguments.TsCreateArgs;
import io.lettuce.core.timeseries.arguments.TsGetArgs;
import io.lettuce.core.timeseries.arguments.TsIncrByArgs;
import io.lettuce.core.timeseries.arguments.TsMGetArgs;
import io.lettuce.core.tracing.TraceContext;
import io.lettuce.core.tracing.TraceContextProvider;
import io.lettuce.core.vector.RawVector;
Expand Down Expand Up @@ -123,14 +133,14 @@
* @author dae won
* @since 4.0
*/
public abstract class AbstractRedisReactiveCommands<K, V>
implements RedisAclReactiveCommands<K, V>, RedisHashReactiveCommands<K, V>, RedisKeyReactiveCommands<K, V>,
RedisStringReactiveCommands<K, V>, RedisListReactiveCommands<K, V>, RedisSetReactiveCommands<K, V>,
RedisSortedSetReactiveCommands<K, V>, RedisScriptingReactiveCommands<K, V>, RedisServerReactiveCommands<K, V>,
RedisHLLReactiveCommands<K, V>, BaseRedisReactiveCommands<K, V>, RedisTransactionalReactiveCommands<K, V>,
RedisGeoReactiveCommands<K, V>, RedisClusterReactiveCommands<K, V>, RedisJsonReactiveCommands<K, V>,
RedisVectorSetReactiveCommands<K, V>, RediSearchReactiveCommands<K, V>, RedisArrayReactiveCommands<K, V>,
RedisBloomFilterReactiveCommands<K, V>, RedisCuckooFilterReactiveCommands<K, V>, RedisTopKReactiveCommands<K, V> {
public abstract class AbstractRedisReactiveCommands<K, V> implements RedisAclReactiveCommands<K, V>,
RedisHashReactiveCommands<K, V>, RedisKeyReactiveCommands<K, V>, RedisStringReactiveCommands<K, V>,
RedisListReactiveCommands<K, V>, RedisSetReactiveCommands<K, V>, RedisSortedSetReactiveCommands<K, V>,
RedisScriptingReactiveCommands<K, V>, RedisServerReactiveCommands<K, V>, RedisHLLReactiveCommands<K, V>,
BaseRedisReactiveCommands<K, V>, RedisTransactionalReactiveCommands<K, V>, RedisGeoReactiveCommands<K, V>,
RedisClusterReactiveCommands<K, V>, RedisJsonReactiveCommands<K, V>, RedisVectorSetReactiveCommands<K, V>,
RediSearchReactiveCommands<K, V>, RedisArrayReactiveCommands<K, V>, RedisBloomFilterReactiveCommands<K, V>,
RedisCuckooFilterReactiveCommands<K, V>, RedisTopKReactiveCommands<K, V>, RedisTimeSeriesReactiveCommands<K, V> {

private final StatefulConnection<K, V> connection;

Expand All @@ -150,6 +160,8 @@ public abstract class AbstractRedisReactiveCommands<K, V>

private final RedisTopKCommandBuilder<K, V> topKCommandBuilder;

private final RedisTimeSeriesCommandBuilder<K, V> timeSeriesCommandBuilder;

private final Supplier<JsonParser> parser;

private final ClientResources clientResources;
Expand Down Expand Up @@ -177,6 +189,7 @@ public AbstractRedisReactiveCommands(StatefulConnection<K, V> connection, RedisC
this.bloomFilterCommandBuilder = new RedisBloomFilterCommandBuilder<>(codec);
this.cuckooFilterCommandBuilder = new RedisCuckooFilterCommandBuilder<>(codec);
this.topKCommandBuilder = new RedisTopKCommandBuilder<>(codec);
this.timeSeriesCommandBuilder = new RedisTimeSeriesCommandBuilder<>(codec);
this.clientResources = connection.getResources();
this.tracingEnabled = clientResources.tracing().isEnabled();
}
Expand Down Expand Up @@ -4575,4 +4588,138 @@ public Mono<String> topKReserve(K key, long k, TopKReserveArgs args) {
return createMono(() -> topKCommandBuilder.topKReserve(key, k, args));
}

// --- Redis Time Series Commands ---

@Override
public Mono<String> tsCreate(K key) {
return createMono(() -> timeSeriesCommandBuilder.tsCreate(key));
}

@Override
public Mono<String> tsCreate(K key, TsCreateArgs createArgs) {
return createMono(() -> timeSeriesCommandBuilder.tsCreate(key, createArgs));
}

@Override
public Mono<String> tsAlter(K key, TsAlterArgs alterArgs) {
return createMono(() -> timeSeriesCommandBuilder.tsAlter(key, alterArgs));
}

@Override
public Mono<String> tsCreateRule(K sourceKey, K destKey, TsAggregationType aggregationType, long bucketDuration) {
return createMono(() -> timeSeriesCommandBuilder.tsCreateRule(sourceKey, destKey, aggregationType, bucketDuration));
}

@Override
public Mono<String> tsCreateRule(K sourceKey, K destKey, TsAggregationType aggregationType, long bucketDuration,
long alignTimestamp) {
return createMono(() -> timeSeriesCommandBuilder.tsCreateRule(sourceKey, destKey, aggregationType, bucketDuration,
alignTimestamp));
}

@Override
public Mono<String> tsDeleteRule(K sourceKey, K destKey) {
return createMono(() -> timeSeriesCommandBuilder.tsDeleteRule(sourceKey, destKey));
}

@Override
public Mono<Long> tsDel(K key, long fromTimestamp, long toTimestamp) {
return createMono(() -> timeSeriesCommandBuilder.tsDel(key, fromTimestamp, toTimestamp));
}

@Override
public Mono<Long> tsAdd(K key, long timestamp, double value) {
return createMono(() -> timeSeriesCommandBuilder.tsAdd(key, timestamp, value));
}

@Override
public Mono<Long> tsAdd(K key, long timestamp, double value, TsAddArgs addArgs) {
return createMono(() -> timeSeriesCommandBuilder.tsAdd(key, timestamp, value, addArgs));
}

@Override
public Mono<Long> tsAdd(K key, double value) {
return createMono(() -> timeSeriesCommandBuilder.tsAdd(key, value));
}

@Override
public Flux<Long> tsMAdd(Map.Entry<K, TsSample>... entries) {
return createDissolvingFlux(() -> timeSeriesCommandBuilder.tsMAdd(entries));
}

@Override
public Flux<Long> tsMAdd(Map.Entry<K, TsSample> entry) {
return createDissolvingFlux(() -> timeSeriesCommandBuilder.tsMAdd(entry));
}

@Override
public Mono<Long> tsIncrBy(K key, double value) {
return createMono(() -> timeSeriesCommandBuilder.tsIncrBy(key, value));
}

@Override
public Mono<Long> tsIncrBy(K key, double value, TsIncrByArgs incrByArgs) {
return createMono(() -> timeSeriesCommandBuilder.tsIncrBy(key, value, incrByArgs));
}

@Override
public Mono<Long> tsDecrBy(K key, double value) {
return createMono(() -> timeSeriesCommandBuilder.tsDecrBy(key, value));
}

@Override
public Mono<Long> tsDecrBy(K key, double value, TsIncrByArgs decrByArgs) {
return createMono(() -> timeSeriesCommandBuilder.tsDecrBy(key, value, decrByArgs));
}

@Override
public Mono<TsSample> tsGet(K key) {
return createMono(() -> timeSeriesCommandBuilder.tsGet(key));
}

@Override
public Mono<TsSample> tsGet(K key, TsGetArgs getArgs) {
return createMono(() -> timeSeriesCommandBuilder.tsGet(key, getArgs));
}

@Override
public Flux<TsMGetValue<K>> tsMGet(V... filters) {
return createDissolvingFlux(() -> timeSeriesCommandBuilder.tsMGet(filters));
}

@Override
public Flux<TsMGetValue<K>> tsMGet(V filter) {
return createDissolvingFlux(() -> timeSeriesCommandBuilder.tsMGet(filter));
}

@Override
public Flux<TsMGetValue<K>> tsMGet(TsMGetArgs mGetArgs, V... filters) {
return createDissolvingFlux(() -> timeSeriesCommandBuilder.tsMGet(mGetArgs, filters));
}

@Override
public Flux<TsMGetValue<K>> tsMGet(TsMGetArgs mGetArgs, V filter) {
return createDissolvingFlux(() -> timeSeriesCommandBuilder.tsMGet(mGetArgs, filter));
}

@Override
public Mono<TsInfoValue<K>> tsInfo(K key) {
return createMono(() -> timeSeriesCommandBuilder.tsInfo(key));
}

@Override
public Mono<TsInfoValue<K>> tsInfoDebug(K key) {
return createMono(() -> timeSeriesCommandBuilder.tsInfoDebug(key));
}

@Override
public Flux<K> tsQueryIndex(V... filters) {
return createDissolvingFlux(() -> timeSeriesCommandBuilder.tsQueryIndex(filters));
}

@Override
public Flux<K> tsQueryIndex(V filter) {
return createDissolvingFlux(() -> timeSeriesCommandBuilder.tsQueryIndex(filter));
}

}
Loading
Loading