diff --git a/cspell.config.yaml b/cspell.config.yaml index e1ffc54e..8829ccff 100644 --- a/cspell.config.yaml +++ b/cspell.config.yaml @@ -89,6 +89,7 @@ words: - propertylistserialization - pubspec - pwsh + - quantiles - rdata - reactivecircus # From .github dir, doesn't show up in "**" check? - readlink @@ -110,6 +111,7 @@ words: - subosito # From .github dir, doesn't show up in "**" check? - swiftshader # From .github dir, doesn't show up in "**" check? - sysroot + - tdigest - temurin # From .github dir, doesn't show up in "**" check? - udevadm # From .github/workflows/e2e.yaml - udid # Unique Device Identifier diff --git a/packages/redis_client/lib/src/redis_client.dart b/packages/redis_client/lib/src/redis_client.dart index f6a05c7a..c2137a45 100644 --- a/packages/redis_client/lib/src/redis_client.dart +++ b/packages/redis_client/lib/src/redis_client.dart @@ -160,6 +160,9 @@ class RedisClient { /// The Redis Time Series commands. RedisTimeSeries get timeSeries => RedisTimeSeries._(client: this); + /// The Redis T-Digest commands. + RedisTDigest get tdigest => RedisTDigest._(client: this); + /// Authenticate to the Redis server. /// Equivalent to the `AUTH` command. /// https://redis.io/commands/auth @@ -889,6 +892,70 @@ class RedisTimeSeries { } } +/// {@template redis_t_digest} +/// A client for interacting with the Redis T-Digest data type. +/// See https://redis.io/docs/latest/develop/data-types/probabilistic/t-digest +/// {@endtemplate} +class RedisTDigest { + /// {@macro redis_t_digest} + const RedisTDigest._({required RedisClient client}) : _client = client; + + final RedisClient _client; + + /// Create a new T-Digest. + /// Equivalent to the `TDIGEST.CREATE` command. + /// https://redis.io/commands/tdigest.create + Future create({required String key, required int compression}) { + return _client.execute(['TDIGEST.CREATE', key, 'COMPRESSION', compression]); + } + + /// Add one or more [observations] to the T-Digest specified by [key]. + /// Equivalent to the `TDIGEST.ADD` command. + /// https://redis.io/commands/tdigest.add + Future add({ + required String key, + required List observations, + }) { + return _client.execute([ + 'TDIGEST.ADD', + key, + ...observations.map((observation) => observation.toString()), + ]); + } + + /// Reset the T-Digest specified by [key]. + /// Equivalent to the `TDIGEST.RESET` command. + /// https://redis.io/commands/tdigest.reset + Future reset({required String key}) { + return _client.execute(['TDIGEST.RESET', key]); + } + + /// Compute the [quantiles] of the T-Digest specified by [key]. + /// Returns a list of [quantiles] in the same order as the [quantiles] list. + /// If the T-Digest is empty, the returned list will contain `null` values. + /// Equivalent to the `TDIGEST.QUANTILE` command. + /// https://redis.io/commands/tdigest.quantile + Future> quantile({ + required String key, + required List quantiles, + }) async { + final results = + await _client.execute([ + 'TDIGEST.QUANTILE', + key, + ...quantiles.map((quantile) => quantile.toString()), + ]) + as List; + + return results.map((result) { + if (result is RespBulkString && result.payload != null) { + return double.tryParse(result.payload!); + } + return null; + }).toList(); + } +} + extension on RedisSocketOptions { /// The connection URI for the Redis server derived from the socket options. Uri get connectionUri => Uri.parse('redis://$host:$port'); diff --git a/packages/redis_client/test/src/redis_client_test.dart b/packages/redis_client/test/src/redis_client_test.dart index 3cca1fab..0867aeed 100644 --- a/packages/redis_client/test/src/redis_client_test.dart +++ b/packages/redis_client/test/src/redis_client_test.dart @@ -664,5 +664,43 @@ void main() { }); }); }); + + group('TDIGEST', () { + setUp(() async { + await client.connect(); + }); + + tearDown(() async { + try { + await client.execute(['FLUSHALL SYNC']); + } on Exception { + // ignore + } + }); + + group('CREATE/ADD/RESET/QUANTILE', () { + const key = 't-digest'; + + test('computes quantiles', () async { + await expectLater( + client.tdigest.create(key: key, compression: 100), + completes, + ); + await expectLater( + client.tdigest.add(key: key, observations: [1, 2, 3]), + completes, + ); + await expectLater( + client.tdigest.quantile(key: key, quantiles: [0.5, 0.9]), + completion(equals([2.0, 3.0])), + ); + await expectLater(client.tdigest.reset(key: key), completes); + await expectLater( + client.tdigest.quantile(key: key, quantiles: [0.5, 0.9]), + completion(equals([null, null])), + ); + }); + }); + }); }); }