feat(redis_client): add support for t-digest functions (#3150)

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

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<void> 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<void> add({
+    required String key,
+    required List<double> 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<void> 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<List<double?>> quantile({
+    required String key,
+    required List<double> quantiles,
+  }) async {
+    final results =
+        await _client.execute([
+              'TDIGEST.QUANTILE',
+              key,
+              ...quantiles.map((quantile) => quantile.toString()),
+            ])
+            as List<RespType>;
+
+    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');

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])),
+          );
+        });
+      });
+    });
   });
 }