Add HyperLogLog Commands (#697)

* add hll commands: pfadd, pfcount, pfmerge

* convert tests for bun

* type fixes

* DX-21: pipeline & redis exports

* DX-21: fix pipeline

pkg/commands/mod.ts

@@ -86,6 +86,9 @@ export * from "./msetnx";
 export * from "./persist";
 export * from "./pexpire";
 export * from "./pexpireat";
+export * from './pfadd';
+export * from './pfcount';
+export * from './pfmerge';
 export * from "./ping";
 export * from "./psetex";
 export * from "./pttl";

pkg/commands/pfadd.test.ts

@@ -0,0 +1,97 @@
+import { newHttpClient, randomID, keygen } from "../test-utils.ts";
+
+import { afterEach, describe, expect, test } from "bun:test";
+
+import { PfAddCommand } from "./pfadd.ts";
+import { PfCountCommand } from "./pfcount.ts";
+import { PfMergeCommand } from "./pfmerge.ts";
+
+const client = newHttpClient();
+
+const { newKey, cleanup } = keygen();
+afterEach(cleanup);
+
+describe("adding multiple elements at once", () => {
+  const key = newKey();
+  test("returns 1 if successful, returns 3 as the cardinality", async () => {
+    const value1 = randomID();
+    const value2 = randomID();
+    const value3 = randomID();
+
+    const res = await new PfAddCommand([key, value1, value2, value3]).exec(
+      client
+    );
+    expect(res).toBe(1);
+
+    const res2 = await new PfCountCommand([key]).exec(client);
+
+    expect(res2).toBe(3);
+  });
+});
+
+describe("inserting the same element multiple times", () => {
+  const key = newKey();
+  const value1 = randomID();
+  const value2 = randomID();
+
+  test("modified succesfully and returned correct cardinality for repeated elements", async () => {
+    const resInsert = await new PfAddCommand([
+      key,
+      value1,
+      value1,
+      value2,
+      value2,
+    ]).exec(client);
+    expect(resInsert).toBe(1);
+
+    const resCount = await new PfCountCommand([key]).exec(client);
+    expect(resCount).toBe(2);
+  });
+});
+
+describe("adding the same strings on different lines doesn't modify the HLL", () => {
+  const key = newKey();
+
+  const value1 = randomID();
+  const value2 = randomID();
+  const value3 = randomID();
+
+  test("modifies the HLL on the first insertion of strings", async () => {
+    const resAdd = await new PfAddCommand([key, value1, value2, value3]).exec(
+      client
+    );
+    expect(resAdd).toBe(1);
+
+    const resAddDuplicate = await new PfAddCommand([
+      key,
+      value1,
+      value2,
+      value3,
+    ]).exec(client);
+    expect(resAddDuplicate).toBe(0);
+  });
+});
+
+describe("merge HLLs with overlapping values and count", () => {
+  const key1 = newKey();
+  const key2 = newKey();
+  const mergedKey = newKey();
+  const value1 = randomID();
+  const value2 = randomID();
+  const value3 = randomID();
+  const value4 = randomID();
+
+  test("insert overlapping strings into two HLLs", async () => {
+    await new PfAddCommand([key1, value1, value2, value3]).exec(client);
+    const resAdd = await new PfAddCommand([key2, value3, value4]).exec(client);
+    expect(resAdd).toBe(1);
+
+    const resMerge = await new PfMergeCommand([mergedKey, key1, key2]).exec(
+      client
+    );
+    expect(resMerge).toBe("OK");
+
+    const resCount = await new PfCountCommand([mergedKey]).exec(client);
+    expect(resCount).toBe(4);
+  });
+});

pkg/commands/pfadd.ts

@@ -0,0 +1,13 @@
+import { Command, CommandOptions } from "./command.ts";
+
+/**
+ * @see https://redis.io/commands/pfadd
+ */
+export class PfAddCommand<TData = string> extends Command<number, number> {
+  constructor(
+    cmd: [string, ...(TData[] | TData[])],
+    opts?: CommandOptions<number, number>
+  ) {
+    super(["pfadd", ...cmd], opts);
+  }
+}

pkg/commands/pfcount.test.ts

@@ -0,0 +1,56 @@
+import { newHttpClient, keygen, randomID } from "../test-utils.ts";
+import { afterEach, expect, test, describe } from "bun:test";
+
+import { PfAddCommand } from "./pfadd.ts";
+import { PfCountCommand } from "./pfcount.ts";
+
+const client = newHttpClient();
+
+const { newKey, cleanup } = keygen();
+afterEach(cleanup);
+
+describe("simple cardinality check", () => {
+  const key = newKey();
+
+  test("insert multiple unique strings", async () => {
+    const value1 = randomID();
+    const value2 = randomID();
+    const value3 = randomID();
+    await new PfAddCommand([key, value1, value2, value3]).exec(client);
+
+    const resCount = await new PfCountCommand([key]).exec(client);
+    expect(resCount).toBe(3);
+  });
+});
+
+describe("multiple keys cardinality check", () => {
+  const key1 = newKey();
+  const key2 = newKey();
+  const value1 = randomID();
+  const value2 = randomID();
+  const value3 = randomID();
+  const value4 = randomID();
+  const value5 = randomID();
+
+  test("insert unique strings into two HLLs", async () => {
+    await new PfAddCommand([key1, value1, value2]).exec(client);
+    await new PfAddCommand([key2, value3, value4]).exec(client);
+
+    const resCount = await new PfCountCommand([key1, key2]).exec(client);
+    expect(resCount).toBe(4);
+  });
+});
+
+describe("cardinality after repeated insertions", () => {
+  const key = newKey();
+  const value1 = randomID();
+  const value2 = randomID();
+
+  test("insert strings and then re-insert them", async () => {
+    await new PfAddCommand([key, value1, value2]).exec(client);
+    await new PfAddCommand([key, value1, value2]).exec(client);
+
+    const resCount = await new PfCountCommand([key]).exec(client);
+    expect(resCount).toBe(2);
+  });
+});

pkg/commands/pfcount.ts

@@ -0,0 +1,13 @@
+import { Command, CommandOptions } from "./command.ts";
+
+/**
+ * @see https://redis.io/commands/pfcount
+ */
+export class PfCountCommand extends Command<number, number> {
+  constructor(
+    cmd: [string, ...(string[] | string[])],
+    opts?: CommandOptions<number, number>
+  ) {
+    super(["pfcount", ...cmd], opts);
+  }
+}

pkg/commands/pfmerge.test.ts

@@ -0,0 +1,75 @@
+import { newHttpClient, randomID, keygen } from "../test-utils.ts";
+
+import { afterEach, expect, test, describe } from "bun:test";
+
+import { PfAddCommand } from "./pfadd.ts";
+import { PfCountCommand } from "./pfcount.ts";
+import { PfMergeCommand } from "./pfmerge.ts";
+
+const client = newHttpClient();
+
+const { newKey, cleanup } = keygen();
+
+afterEach(cleanup);
+
+describe("merge HLLs with distinct values and count", () => {
+  const key1 = newKey();
+  const key2 = newKey();
+  const mergedKey = newKey();
+  const value1 = randomID();
+  const value2 = randomID();
+  const value3 = randomID();
+  const value4 = randomID();
+
+  test("insert distinct strings into two HLLs", async () => {
+    await new PfAddCommand([key1, value1, value2]).exec(client);
+    const resAdd = await new PfAddCommand([key2, value3, value4]).exec(client);
+    expect(resAdd).toBe(1);
+
+    const resMerge = await new PfMergeCommand([mergedKey, key1, key2]).exec(
+      client
+    );
+    expect(resMerge).toBe("OK");
+
+    const resCount = await new PfCountCommand([mergedKey]).exec(client);
+    expect(resCount).toBe(4);
+  });
+});
+
+describe("merge HLL with an empty HLL", () => {
+  const key = newKey();
+  const emptyKey = newKey();
+  const mergedKey = newKey();
+  const value1 = randomID();
+
+  test("insert a string into an HLL and keep another HLL empty", async () => {
+    const resAdd = await new PfAddCommand([key, value1]).exec(client);
+    expect(resAdd).toBe(1);
+
+    const resMerge = await new PfMergeCommand([mergedKey, key, emptyKey]).exec(
+      client
+    );
+    expect(resMerge).toBe("OK");
+
+    const resCount = await new PfCountCommand([mergedKey]).exec(client);
+    expect(resCount).toBe(1);
+  });
+});
+
+describe("merge two empty HLLs", () => {
+  const emptyKey1 = newKey();
+  const emptyKey2 = newKey();
+  const mergedKey = newKey();
+
+  test("merge two empty HLLs", async () => {
+    const resMerge = await new PfMergeCommand([
+      mergedKey,
+      emptyKey1,
+      emptyKey2,
+    ]).exec(client);
+    expect(resMerge).toBe("OK");
+
+    const resCount = await new PfCountCommand([mergedKey]).exec(client);
+    expect(resCount).toBe(0);
+  });
+});

pkg/commands/pfmerge.ts

@@ -0,0 +1,13 @@
+import { Command, CommandOptions } from "./command.ts";
+
+/**
+ * @see https://redis.io/commands/pfmerge
+ */
+export class PfMergeCommand extends Command<"OK", "OK"> {
+  constructor(
+    cmd: [destination_key: string, ...(string[] | string[])],
+    opts?: CommandOptions<"OK", "OK">
+  ) {
+    super(["pfmerge", ...cmd], opts);
+  }
+}

pkg/pipeline.ts

@@ -88,6 +88,9 @@ import {
   PExpireCommand,
   PSetEXCommand,
   PTtlCommand,
+  PfAddCommand,
+  PfCountCommand,
+  PfMergeCommand,
   PersistCommand,
   PingCommand,
   PublishCommand,
@@ -657,6 +660,24 @@ export class Pipeline<TCommands extends Command<any, any>[] = []> {
   pexpireat = (...args: CommandArgs<typeof PExpireAtCommand>) =>
     this.chain(new PExpireAtCommand(args, this.commandOptions));
 
+    /**
+   * @see https://redis.io/commands/pfadd
+   */
+  pfadd = (...args: CommandArgs<typeof PfAddCommand>) =>
+    this.chain(new PfAddCommand(args, this.commandOptions));
+
+  /**
+   * @see https://redis.io/commands/pfcount
+   */
+  pfcount = (...args: CommandArgs<typeof PfCountCommand>) =>
+    this.chain(new PfCountCommand(args, this.commandOptions));
+
+  /**
+   * @see https://redis.io/commands/pfmerge
+   */
+  pfmerge = (...args: CommandArgs<typeof PfMergeCommand>) =>
+    this.chain(new PfMergeCommand(args, this.commandOptions));
+
   /**
    * @see https://redis.io/commands/ping
    */

pkg/redis.ts

@@ -88,6 +88,9 @@ import {
   PExpireCommand,
   PSetEXCommand,
   PTtlCommand,
+  PfAddCommand,
+  PfCountCommand,
+  PfMergeCommand,
   PersistCommand,
   PingCommand,
   PublishCommand,
@@ -803,6 +806,24 @@ export class Redis {
   pexpireat = (...args: CommandArgs<typeof PExpireAtCommand>) =>
     new PExpireAtCommand(args, this.opts).exec(this.client);
 
+    /**
+   * @see https://redis.io/commands/pfadd
+   */
+    pfadd = (...args: CommandArgs<typeof PfAddCommand>) =>
+    new PfAddCommand(args, this.opts).exec(this.client);
+
+  /**
+   * @see https://redis.io/commands/pfcount
+   */
+  pfcount = (...args: CommandArgs<typeof PfCountCommand>) =>
+    new PfCountCommand(args, this.opts).exec(this.client);
+
+  /**
+   * @see https://redis.io/commands/pfmerge
+   */
+  pfmerge = (...args: CommandArgs<typeof PfMergeCommand>) =>
+    new PfMergeCommand(args, this.opts).exec(this.client);
+
   /**
    * @see https://redis.io/commands/ping
    */