From bbd1754fce84b4c2563c3118f9fa24b5b929ca5e Mon Sep 17 00:00:00 2001 From: Felix Angelov Date: Wed, 2 Apr 2025 16:38:56 -0500 Subject: [PATCH] feat(redis_client): add `TS.RANGE` (#3037) --- .../lib/shorebird_redis_client.dart | 7 +- .../redis_client/lib/src/redis_client.dart | 261 +++++++++++++++--- .../test/src/redis_client_test.dart | 148 ++++++++++ 3 files changed, 380 insertions(+), 36 deletions(-) diff --git a/packages/redis_client/lib/shorebird_redis_client.dart b/packages/redis_client/lib/shorebird_redis_client.dart index f22cd637..a968184d 100644 --- a/packages/redis_client/lib/shorebird_redis_client.dart +++ b/packages/redis_client/lib/shorebird_redis_client.dart @@ -7,9 +7,14 @@ export 'src/redis_client.dart' RedisLogger, RedisSocketOptions, RedisTimeSeries, + RedisTimeSeriesAggregation, + RedisTimeSeriesAggregator, + RedisTimeSeriesAlign, RedisTimeSeriesClientClock, RedisTimeSeriesClock, RedisTimeSeriesDuplicatePolicy, RedisTimeSeriesEncoding, + RedisTimeSeriesFromTimestamp, RedisTimeSeriesServerClock, - RedisTimeSeriesTimestamp; + RedisTimeSeriesTimestamp, + RedisTimeSeriesToTimestamp; diff --git a/packages/redis_client/lib/src/redis_client.dart b/packages/redis_client/lib/src/redis_client.dart index 9c56388b..13421809 100644 --- a/packages/redis_client/lib/src/redis_client.dart +++ b/packages/redis_client/lib/src/redis_client.dart @@ -515,41 +515,6 @@ enum RedisTimeSeriesDuplicatePolicy { String toArgument() => name.toUpperCase(); } -/// {@template redis_time_series_clock} -/// An abstract class which represents the clock -/// in the context of a redis time series instance. -/// {@endtemplate} -// ignore: one_member_abstracts -abstract class RedisTimeSeriesClock { - /// {@macro redis_time_series_clock} - const RedisTimeSeriesClock(); - - /// Returns a timestamp representing the current moment. - RedisTimeSeriesTimestamp now(); -} - -/// {@template redis_time_series_server_clock} -/// A [RedisTimeSeriesClock] that represents time on the server. -/// {@endtemplate} -class RedisTimeSeriesServerClock extends RedisTimeSeriesClock { - /// {@macro redis_time_series_server_clock} - const RedisTimeSeriesServerClock(); - @override - RedisTimeSeriesTimestamp now() => const RedisTimeSeriesTimestamp._('*'); -} - -/// {@template redis_time_series_client_clock} -/// A [RedisTimeSeriesClock] that represents time on the client. -/// {@endtemplate} -class RedisTimeSeriesClientClock extends RedisTimeSeriesClock { - /// {@macro redis_time_series_client_clock} - const RedisTimeSeriesClientClock(); - @override - RedisTimeSeriesTimestamp now() { - return RedisTimeSeriesTimestamp(DateTime.timestamp()); - } -} - /// {@template redis_time_series_timestamp} /// is Unix time (integer, in milliseconds) specifying the sample timestamp or * /// to set the sample timestamp to the Unix time of the server's clock. @@ -596,6 +561,181 @@ class RedisTimeSeriesTimestamp { final String value; } +/// {@template redis_time_series_clock} +/// An abstract class which represents the clock +/// in the context of a redis time series instance. +/// {@endtemplate} +// ignore: one_member_abstracts +abstract class RedisTimeSeriesClock { + /// {@macro redis_time_series_clock} + const RedisTimeSeriesClock(); + + /// Returns a timestamp representing the current moment. + RedisTimeSeriesTimestamp now(); +} + +/// {@template redis_time_series_server_clock} +/// A [RedisTimeSeriesClock] that represents time on the server. +/// {@endtemplate} +class RedisTimeSeriesServerClock extends RedisTimeSeriesClock { + /// {@macro redis_time_series_server_clock} + const RedisTimeSeriesServerClock(); + @override + RedisTimeSeriesTimestamp now() => const RedisTimeSeriesTimestamp._('*'); +} + +/// {@template redis_time_series_client_clock} +/// A [RedisTimeSeriesClock] that represents time on the client. +/// {@endtemplate} +class RedisTimeSeriesClientClock extends RedisTimeSeriesClock { + /// {@macro redis_time_series_client_clock} + const RedisTimeSeriesClientClock(); + @override + RedisTimeSeriesTimestamp now() { + return RedisTimeSeriesTimestamp(DateTime.timestamp()); + } +} + +/// {@template redis_time_series_from_timestamp} +/// The start timestamp for the range query (integer Unix timestamp in +/// milliseconds). +/// {@endtemplate} +class RedisTimeSeriesFromTimestamp { + /// {@macro redis_time_series_from_timestamp} + RedisTimeSeriesFromTimestamp(DateTime dateTime) + : this._('${dateTime.millisecondsSinceEpoch}'); + + /// The timestamp of the earliest sample among all the time series + /// that passes the provided filter. + const RedisTimeSeriesFromTimestamp.start() : value = '-'; + + const RedisTimeSeriesFromTimestamp._(this.value); + + /// The underlying value of the timestamp. + final String value; +} + +/// {@template redis_time_series_to_timestamp} +/// The end timestamp for the range query (integer Unix timestamp in +/// milliseconds). +/// {@endtemplate} +class RedisTimeSeriesToTimestamp { + /// {@macro redis_time_series_to_timestamp} + RedisTimeSeriesToTimestamp(DateTime dateTime) + : this._('${dateTime.millisecondsSinceEpoch}'); + + /// The timestamp of the latest sample among all the time series that passes + /// the provided filter. + const RedisTimeSeriesToTimestamp.end() : value = '+'; + + const RedisTimeSeriesToTimestamp._(this.value); + + /// The underlying value of the timestamp. + final String value; +} + +/// {@template redis_time_series_align} +/// The time bucket alignment control for AGGREGATION. It controls the time +/// bucket timestamps by changing the reference timestamp on which a bucket is +/// defined. +/// {@endtemplate} +class RedisTimeSeriesAlign { + /// A specific timestamp: align the reference timestamp to a specific time. + RedisTimeSeriesAlign(DateTime date) + : this._('${date.millisecondsSinceEpoch}'); + + /// The reference timestamp will be the query start interval time + /// (fromTimestamp) which can't be -. + RedisTimeSeriesAlign.start() : this._('-'); + + /// The reference timestamp will be the query end interval time (toTimestamp) + /// which can't be +. + RedisTimeSeriesAlign.end() : this._('+'); + + RedisTimeSeriesAlign._(this.value); + + /// The underlying value of the alignment. + final String value; +} + +enum RedisTimeSeriesAggregator { + /// Arithmetic mean of all values + average, + + /// Sum of all values + sum, + + /// Minimum value + min, + + /// Maximum value + max, + + /// Difference between the maximum and the minimum value + range, + + /// Number of values + count, + + /// Value with lowest timestamp in the bucket + first, + + /// Value with highest timestamp in the bucket + last, + + /// Population standard deviation of the values + populationStandardDeviation, + + /// Sample standard deviation of the values + sampleStandardDeviation, + + /// Population variance of the values + populationVariance, + + /// Sample variance of the values + sampleVariance, + + /// Time-weighted average over the bucket's timeframe + timeWeightedAverage; + + /// Converts the enum to an argument that can be passed directly to + /// `execute`. + String toArgument() { + return switch (this) { + RedisTimeSeriesAggregator.average => 'avg', + RedisTimeSeriesAggregator.sum => 'sum', + RedisTimeSeriesAggregator.min => 'min', + RedisTimeSeriesAggregator.max => 'max', + RedisTimeSeriesAggregator.range => 'range', + RedisTimeSeriesAggregator.count => 'count', + RedisTimeSeriesAggregator.first => 'first', + RedisTimeSeriesAggregator.last => 'last', + RedisTimeSeriesAggregator.populationStandardDeviation => 'std.p', + RedisTimeSeriesAggregator.sampleStandardDeviation => 'std.s', + RedisTimeSeriesAggregator.populationVariance => 'var.p', + RedisTimeSeriesAggregator.sampleVariance => 'var.s', + RedisTimeSeriesAggregator.timeWeightedAverage => 'twa', + }; + } +} + +/// {@template redis_time_series_aggregation} +/// Aggregates time series samples into time buckets. +/// {@endtemplate} +class RedisTimeSeriesAggregation { + /// {@macro redis_time_series_aggregation} + const RedisTimeSeriesAggregation({ + required this.aggregator, + required this.bucketDuration, + }); + + /// The aggregation type. + final RedisTimeSeriesAggregator aggregator; + + /// The duration of each bucket. + final Duration bucketDuration; +} + /// {@template redis_time_series} /// An object that adds support for storing and querying timestamped data /// points. @@ -692,6 +832,57 @@ class RedisTimeSeries { value: double.parse(value.payload), ); } + + /// Query a range in forward direction. + /// Equivalent to the `TS.RANGE` command. + /// https://redis.io/commands/ts.range + Future> range({ + required String key, + required RedisTimeSeriesFromTimestamp from, + required RedisTimeSeriesToTimestamp to, + List? filterByTimestamp, + ({double min, double max})? filterByValue, + int? count, + RedisTimeSeriesAlign? align, + RedisTimeSeriesAggregation? aggregation, + }) async { + final results = + await _client.execute([ + 'TS.RANGE', + key, + from.value, + to.value, + if (filterByTimestamp != null) ...[ + 'FILTER_BY_TS', + ...filterByTimestamp.map((t) => t.value), + ], + if (filterByValue != null) ...[ + 'FILTER_BY_VALUE', + filterByValue.min, + filterByValue.max, + ], + if (count != null) ...['COUNT', count], + if (align != null) ...['ALIGN', align.value], + if (aggregation != null) ...[ + 'AGGREGATION', + aggregation.aggregator.toArgument(), + aggregation.bucketDuration.inMilliseconds, + ], + ]) + as List; + return results.map((result) { + final payload = result.payload as List; + final timestamp = payload[0] as RespInteger; + final value = payload[1] as RespSimpleString; + return ( + timestamp: DateTime.fromMillisecondsSinceEpoch( + timestamp.payload, + isUtc: true, + ), + value: double.parse(value.payload), + ); + }).toList(); + } } extension on RedisSocketOptions { diff --git a/packages/redis_client/test/src/redis_client_test.dart b/packages/redis_client/test/src/redis_client_test.dart index 6c4a1a17..8706a13b 100644 --- a/packages/redis_client/test/src/redis_client_test.dart +++ b/packages/redis_client/test/src/redis_client_test.dart @@ -515,6 +515,154 @@ void main() { ); }); }); + + group('RANGE', () { + const key = 'sensor'; + final data = [ + (timestamp: DateTime(2020).toUtc(), value: 1.0), + (timestamp: DateTime(2021).toUtc(), value: 2.0), + (timestamp: DateTime(2022).toUtc(), value: 3.0), + (timestamp: DateTime(2023).toUtc(), value: 4.0), + ]; + + setUp(() async { + await client.connect(); + await expectLater(client.delete(key: key), completes); + for (final tuple in data) { + await expectLater( + client.timeSeries.add( + key: key, + timestamp: RedisTimeSeriesTimestamp(tuple.timestamp), + value: tuple.value, + ), + completes, + ); + } + }); + + test('completes', () async { + await expectLater( + client.timeSeries.range( + key: key, + from: const RedisTimeSeriesFromTimestamp.start(), + to: const RedisTimeSeriesToTimestamp.end(), + ), + completion(containsAllInOrder(data)), + ); + + await expectLater( + client.timeSeries.range( + key: key, + from: const RedisTimeSeriesFromTimestamp.start(), + to: const RedisTimeSeriesToTimestamp.end(), + count: 3, + ), + completion(containsAllInOrder(data.sublist(0, 2))), + ); + + await expectLater( + client.timeSeries.range( + key: key, + from: const RedisTimeSeriesFromTimestamp.start(), + to: const RedisTimeSeriesToTimestamp.end(), + filterByTimestamp: [ + RedisTimeSeriesTimestamp(DateTime(2023).toUtc()), + ], + ), + completion(containsAllInOrder([data.last])), + ); + + await expectLater( + client.timeSeries.range( + key: key, + from: const RedisTimeSeriesFromTimestamp.start(), + to: const RedisTimeSeriesToTimestamp.end(), + filterByValue: (min: 0, max: 1), + ), + completion(containsAllInOrder(data.sublist(0, 1))), + ); + + await expectLater( + client.timeSeries.range( + key: key, + from: RedisTimeSeriesFromTimestamp(data.first.timestamp), + to: const RedisTimeSeriesToTimestamp.end(), + aggregation: const RedisTimeSeriesAggregation( + aggregator: RedisTimeSeriesAggregator.timeWeightedAverage, + bucketDuration: Duration(days: 365 * 3), + ), + align: RedisTimeSeriesAlign.start(), + ), + completion( + containsAllInOrder([ + (timestamp: DateTime(2020).toUtc(), value: 2.4977181459936197), + ( + timestamp: DateTime(2022, 12, 31).toUtc(), + value: 3.998630136986301, + ), + ]), + ), + ); + + await expectLater( + client.timeSeries.range( + key: key, + from: RedisTimeSeriesFromTimestamp(data.first.timestamp), + to: const RedisTimeSeriesToTimestamp.end(), + aggregation: const RedisTimeSeriesAggregation( + aggregator: RedisTimeSeriesAggregator.count, + bucketDuration: Duration(days: 365 * 3), + ), + align: RedisTimeSeriesAlign.start(), + ), + completion( + containsAllInOrder([ + (timestamp: DateTime(2020).toUtc(), value: 3.0), + (timestamp: DateTime(2022, 12, 31).toUtc(), value: 1.0), + ]), + ), + ); + + await expectLater( + client.timeSeries.range( + key: key, + from: const RedisTimeSeriesFromTimestamp.start(), + to: RedisTimeSeriesToTimestamp(data.last.timestamp), + aggregation: const RedisTimeSeriesAggregation( + aggregator: RedisTimeSeriesAggregator.count, + bucketDuration: Duration(days: 365 * 3), + ), + align: RedisTimeSeriesAlign.end(), + ), + completion( + containsAllInOrder([ + (timestamp: DateTime(2017, 1, 2).toUtc(), value: 1.0), + (timestamp: DateTime(2020, 1, 2).toUtc(), value: 2.0), + (timestamp: DateTime(2023).toUtc(), value: 1.0), + ]), + ), + ); + + await expectLater( + client.timeSeries.range( + key: key, + from: const RedisTimeSeriesFromTimestamp.start(), + to: RedisTimeSeriesToTimestamp(data.last.timestamp), + aggregation: const RedisTimeSeriesAggregation( + aggregator: RedisTimeSeriesAggregator.count, + bucketDuration: Duration(days: 365 * 3), + ), + align: RedisTimeSeriesAlign(DateTime(2022).toUtc()), + ), + completion( + containsAllInOrder([ + (timestamp: DateTime(2019, 1, 2).toUtc(), value: 2.0), + (timestamp: DateTime(2022).toUtc(), value: 2.0), + ]), + ), + ); + }); + }); }); }); }