move development of d2mini to the db monorepo and rename db-ivm (#330)

Co-authored-by: Kyle Mathews <[email protected]>

Author: samwillis

.changeset/slimy-lizards-flash.md

@@ -0,0 +1,10 @@
+---
+"@tanstack/svelte-db": patch
+"@tanstack/react-db": patch
+"@tanstack/solid-db": patch
+"@tanstack/db-ivm": patch
+"@tanstack/vue-db": patch
+"@tanstack/db": patch
+---
+
+We have moved development of the differential dataflow implementation from @electric-sql/d2mini to a new @tanstack/db-ivm package inside the tanstack db monorepo to make development simpler.

package.json

@@ -77,6 +77,7 @@
   ],
   "overrides": {
     "@tanstack/db": "workspace:*",
+    "@tanstack/db-ivm": "workspace:*",
     "@tanstack/react-db": "workspace:*",
     "@tanstack/vue-db": "workspace:*"
   }

packages/db-ivm/README.md

@@ -0,0 +1,58 @@
+# IVM implementation for TanStack DB based on Differential Dataflow
+
+This is an implementation of differential dataflow used by TanStack DB, forked from [@electric-sql/d2ts](https://github.com/electric-sql/d2ts), but simplified and without the complexities of multi-dimensional versioning.
+
+It is used internally by the TanStack DB with the live queries compiled to a D2 graph. This library doesn't depend on the TanStack DB and so could be used in other projects.
+
+The API is almost identical to D2TS, but without the need to specify a version when sending data, or to send a frontier to mark the end of a version.
+
+### Basic Usage
+
+Here's a simple example that demonstrates the core concepts:
+
+```typescript
+import { D2, map, filter, debug, MultiSet } from "@tanstack/db-ivm"
+
+// Create a new D2 graph
+const graph = new D2()
+
+// Create an input stream
+// We can specify the type of the input stream, here we are using number.
+const input = graph.newInput<number>()
+
+// Build a simple pipeline that:
+// 1. Takes numbers as input
+// 2. Adds 5 to each number
+// 3. Filters to keep only even numbers
+// Pipelines can have multiple inputs and outputs.
+const output = input.pipe(
+  map((x) => x + 5),
+  filter((x) => x % 2 === 0),
+  debug("output")
+)
+
+// Finalize the pipeline, after this point we can no longer add operators or
+// inputs
+graph.finalize()
+
+// Send some data
+// Data is sent as a MultiSet, which is a map of values to their multiplicity
+// Here we are sending 3 numbers (1-3), each with a multiplicity of 1
+// The key thing to understand is that the MultiSet represents a *change* to
+// the data, not the data itself. "Inserts" and "Deletes" are represented as
+// an element with a multiplicity of 1 or -1 respectively.
+input.sendData(
+  new MultiSet([
+    [1, 1],
+    [2, 1],
+    [3, 1],
+  ])
+)
+
+// Process the data
+graph.run()
+
+// Output will show:
+// 6 (from 1 + 5)
+// 8 (from 3 + 5)
+```

packages/db-ivm/package.json

@@ -0,0 +1,63 @@
+{
+  "name": "@tanstack/db-ivm",
+  "description": "Incremental View Maintenance for TanStack DB based on Differential Dataflow",
+  "version": "0.0.0",
+  "dependencies": {
+    "fractional-indexing": "^3.2.0",
+    "murmurhash-js": "^1.0.0",
+    "sorted-btree": "^1.8.1"
+  },
+  "devDependencies": {
+    "@types/debug": "^4.1.12",
+    "@types/murmurhash-js": "^1.0.6",
+    "@vitest/coverage-istanbul": "^3.0.9"
+  },
+  "exports": {
+    ".": {
+      "import": {
+        "types": "./dist/esm/index.d.ts",
+        "default": "./dist/esm/index.js"
+      },
+      "require": {
+        "types": "./dist/cjs/index.d.cts",
+        "default": "./dist/cjs/index.cjs"
+      }
+    },
+    "./package.json": "./package.json"
+  },
+  "files": [
+    "dist",
+    "src"
+  ],
+  "main": "dist/cjs/index.cjs",
+  "module": "dist/esm/index.js",
+  "packageManager": "[email protected]",
+  "peerDependencies": {
+    "typescript": ">=4.7"
+  },
+  "author": "Sam Willis",
+  "license": "MIT",
+  "repository": {
+    "type": "git",
+    "url": "https://github.com/TanStack/db.git",
+    "directory": "packages/db-ivm"
+  },
+  "homepage": "https://tanstack.com/db",
+  "keywords": [
+    "electric",
+    "sql",
+    "optimistic",
+    "typescript",
+    "ivm",
+    "differential-dataflow"
+  ],
+  "scripts": {
+    "build": "vite build",
+    "dev": "vite build --watch",
+    "lint": "eslint . --fix",
+    "test": "npx vitest --run"
+  },
+  "sideEffects": false,
+  "type": "module",
+  "types": "dist/esm/index.d.ts"
+}

packages/db-ivm/src/d2.ts

@@ -0,0 +1,150 @@
+import { DifferenceStreamWriter } from "./graph.js"
+import type {
+  BinaryOperator,
+  DifferenceStreamReader,
+  UnaryOperator,
+} from "./graph.js"
+import type { MultiSet, MultiSetArray } from "./multiset.js"
+import type { ID2, IStreamBuilder, PipedOperator } from "./types.js"
+
+export class D2 implements ID2 {
+  #streams: Array<DifferenceStreamReader<any>> = []
+  #operators: Array<UnaryOperator<any> | BinaryOperator<any>> = []
+  #nextOperatorId = 0
+  #finalized = false
+
+  constructor() {}
+
+  #checkNotFinalized(): void {
+    if (this.#finalized) {
+      throw new Error(`Graph already finalized`)
+    }
+  }
+
+  getNextOperatorId(): number {
+    this.#checkNotFinalized()
+    return this.#nextOperatorId++
+  }
+
+  newInput<T>(): RootStreamBuilder<T> {
+    this.#checkNotFinalized()
+    const writer = new DifferenceStreamWriter<T>()
+    // Use the root stream builder that exposes the sendData and sendFrontier methods
+    const streamBuilder = new RootStreamBuilder<T>(this, writer)
+    this.#streams.push(streamBuilder.connectReader())
+    return streamBuilder
+  }
+
+  addOperator(operator: UnaryOperator<any> | BinaryOperator<any>): void {
+    this.#checkNotFinalized()
+    this.#operators.push(operator)
+  }
+
+  addStream(stream: DifferenceStreamReader<any>): void {
+    this.#checkNotFinalized()
+    this.#streams.push(stream)
+  }
+
+  finalize() {
+    this.#checkNotFinalized()
+    this.#finalized = true
+  }
+
+  step(): void {
+    if (!this.#finalized) {
+      throw new Error(`Graph not finalized`)
+    }
+    for (const op of this.#operators) {
+      op.run()
+    }
+  }
+
+  pendingWork(): boolean {
+    return this.#operators.some((op) => op.hasPendingWork())
+  }
+
+  run(): void {
+    while (this.pendingWork()) {
+      this.step()
+    }
+  }
+}
+
+export class StreamBuilder<T> implements IStreamBuilder<T> {
+  #graph: ID2
+  #writer: DifferenceStreamWriter<T>
+
+  constructor(graph: ID2, writer: DifferenceStreamWriter<T>) {
+    this.#graph = graph
+    this.#writer = writer
+  }
+
+  connectReader(): DifferenceStreamReader<T> {
+    return this.#writer.newReader()
+  }
+
+  get writer(): DifferenceStreamWriter<T> {
+    return this.#writer
+  }
+
+  get graph(): ID2 {
+    return this.#graph
+  }
+
+  // Don't judge, this is the only way to type this function.
+  // rxjs has very similar code to type its pipe function
+  // https://github.com/ReactiveX/rxjs/blob/master/packages/rxjs/src/internal/util/pipe.ts
+  // We go to 20 operators deep, because surly that's enough for anyone...
+  // A user can always split the pipe into multiple pipes to get around this.
+  pipe<O>(o1: PipedOperator<T, O>): IStreamBuilder<O>
+  // prettier-ignore
+  pipe<T2, O>(o1: PipedOperator<T, T2>, o2: PipedOperator<T2, O>): IStreamBuilder<O>
+  // prettier-ignore
+  pipe<T2, T3, O>(o1: PipedOperator<T, T2>, o2: PipedOperator<T2, T3>, o3: PipedOperator<T3, O>): IStreamBuilder<O>
+  // prettier-ignore
+  pipe<T2, T3, T4, O>(o1: PipedOperator<T, T2>, o2: PipedOperator<T2, T3>, o3: PipedOperator<T3, T4>, o4: PipedOperator<T4, O>): IStreamBuilder<O>
+  // prettier-ignore
+  pipe<T2, T3, T4, T5, O>(o1: PipedOperator<T, T2>, o2: PipedOperator<T2, T3>, o3: PipedOperator<T3, T4>, o4: PipedOperator<T4, T5>, o5: PipedOperator<T5, O>): IStreamBuilder<O>
+  // prettier-ignore
+  pipe<T2, T3, T4, T5, T6, O>(o1: PipedOperator<T, T2>, o2: PipedOperator<T2, T3>, o3: PipedOperator<T3, T4>, o4: PipedOperator<T4, T5>, o5: PipedOperator<T5, T6>, o6: PipedOperator<T6, O>): IStreamBuilder<O>
+  // prettier-ignore
+  pipe<T2, T3, T4, T5, T6, T7, O>(o1: PipedOperator<T, T2>, o2: PipedOperator<T2, T3>, o3: PipedOperator<T3, T4>, o4: PipedOperator<T4, T5>, o5: PipedOperator<T5, T6>, o6: PipedOperator<T6, T7>, o7: PipedOperator<T7, O>): IStreamBuilder<O>
+  // prettier-ignore
+  pipe<T2, T3, T4, T5, T6, T7, T8, O>(o1: PipedOperator<T, T2>, o2: PipedOperator<T2, T3>, o3: PipedOperator<T3, T4>, o4: PipedOperator<T4, T5>, o5: PipedOperator<T5, T6>, o6: PipedOperator<T6, T7>, o7: PipedOperator<T7, T8>, o8: PipedOperator<T8, O>): IStreamBuilder<O>
+  // prettier-ignore
+  pipe<T2, T3, T4, T5, T6, T7, T8, T9, O>(o1: PipedOperator<T, T2>, o2: PipedOperator<T2, T3>, o3: PipedOperator<T3, T4>, o4: PipedOperator<T4, T5>, o5: PipedOperator<T5, T6>, o6: PipedOperator<T6, T7>, o7: PipedOperator<T7, T8>, o8: PipedOperator<T8, T9>, o9: PipedOperator<T9, O>): IStreamBuilder<O>
+  // prettier-ignore
+  pipe<T2, T3, T4, T5, T6, T7, T8, T9, T10, O>(o1: PipedOperator<T, T2>, o2: PipedOperator<T2, T3>, o3: PipedOperator<T3, T4>, o4: PipedOperator<T4, T5>, o5: PipedOperator<T5, T6>, o6: PipedOperator<T6, T7>, o7: PipedOperator<T7, T8>, o8: PipedOperator<T8, T9>, o9: PipedOperator<T9, T10>, o10: PipedOperator<T10, O>): IStreamBuilder<O>
+  // prettier-ignore
+  pipe<T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, O>(o1: PipedOperator<T, T2>, o2: PipedOperator<T2, T3>, o3: PipedOperator<T3, T4>, o4: PipedOperator<T4, T5>, o5: PipedOperator<T5, T6>, o6: PipedOperator<T6, T7>, o7: PipedOperator<T7, T8>, o8: PipedOperator<T8, T9>, o9: PipedOperator<T9, T10>, o10: PipedOperator<T10, T11>, o11: PipedOperator<T11, O>): IStreamBuilder<O>
+  // prettier-ignore
+  pipe<T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, O>(o1: PipedOperator<T, T2>, o2: PipedOperator<T2, T3>, o3: PipedOperator<T3, T4>, o4: PipedOperator<T4, T5>, o5: PipedOperator<T5, T6>, o6: PipedOperator<T6, T7>, o7: PipedOperator<T7, T8>, o8: PipedOperator<T8, T9>, o9: PipedOperator<T9, T10>, o10: PipedOperator<T10, T11>, o11: PipedOperator<T11, T12>, o12: PipedOperator<T12, O>): IStreamBuilder<O>
+  // prettier-ignore
+  pipe<T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, O>(o1: PipedOperator<T, T2>, o2: PipedOperator<T2, T3>, o3: PipedOperator<T3, T4>, o4: PipedOperator<T4, T5>, o5: PipedOperator<T5, T6>, o6: PipedOperator<T6, T7>, o7: PipedOperator<T7, T8>, o8: PipedOperator<T8, T9>, o9: PipedOperator<T9, T10>, o10: PipedOperator<T10, T11>, o11: PipedOperator<T11, T12>, o12: PipedOperator<T12, T13>, o13: PipedOperator<T13, O>): IStreamBuilder<O>
+  // prettier-ignore
+  pipe<T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, O>(o1: PipedOperator<T, T2>, o2: PipedOperator<T2, T3>, o3: PipedOperator<T3, T4>, o4: PipedOperator<T4, T5>, o5: PipedOperator<T5, T6>, o6: PipedOperator<T6, T7>, o7: PipedOperator<T7, T8>, o8: PipedOperator<T8, T9>, o9: PipedOperator<T9, T10>, o10: PipedOperator<T10, T11>, o11: PipedOperator<T11, T12>, o12: PipedOperator<T12, T13>, o13: PipedOperator<T13, T14>, o14: PipedOperator<T14, O>): IStreamBuilder<O>
+  // prettier-ignore
+  pipe<T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, O>(o1: PipedOperator<T, T2>, o2: PipedOperator<T2, T3>, o3: PipedOperator<T3, T4>, o4: PipedOperator<T4, T5>, o5: PipedOperator<T5, T6>, o6: PipedOperator<T6, T7>, o7: PipedOperator<T7, T8>, o8: PipedOperator<T8, T9>, o9: PipedOperator<T9, T10>, o10: PipedOperator<T10, T11>, o11: PipedOperator<T11, T12>, o12: PipedOperator<T12, T13>, o13: PipedOperator<T13, T14>, o14: PipedOperator<T14, T15>, o15: PipedOperator<T15, O>): IStreamBuilder<O>
+  // prettier-ignore
+  pipe<T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, O>(o1: PipedOperator<T, T2>, o2: PipedOperator<T2, T3>, o3: PipedOperator<T3, T4>, o4: PipedOperator<T4, T5>, o5: PipedOperator<T5, T6>, o6: PipedOperator<T6, T7>, o7: PipedOperator<T7, T8>, o8: PipedOperator<T8, T9>, o9: PipedOperator<T9, T10>, o10: PipedOperator<T10, T11>, o11: PipedOperator<T11, T12>, o12: PipedOperator<T12, T13>, o13: PipedOperator<T13, T14>, o14: PipedOperator<T14, T15>, o15: PipedOperator<T15, T16>, o16: PipedOperator<T16, O>): IStreamBuilder<O>
+  // prettier-ignore
+  pipe<T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, O>(o1: PipedOperator<T, T2>, o2: PipedOperator<T2, T3>, o3: PipedOperator<T3, T4>, o4: PipedOperator<T4, T5>, o5: PipedOperator<T5, T6>, o6: PipedOperator<T6, T7>, o7: PipedOperator<T7, T8>, o8: PipedOperator<T8, T9>, o9: PipedOperator<T9, T10>, o10: PipedOperator<T10, T11>, o11: PipedOperator<T11, T12>, o12: PipedOperator<T12, T13>, o13: PipedOperator<T13, T14>, o14: PipedOperator<T14, T15>, o15: PipedOperator<T15, T16>, o16: PipedOperator<T16, T17>, o17: PipedOperator<T17, O>): IStreamBuilder<O>
+  // prettier-ignore
+  pipe<T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, O>(o1: PipedOperator<T, T2>, o2: PipedOperator<T2, T3>, o3: PipedOperator<T3, T4>, o4: PipedOperator<T4, T5>, o5: PipedOperator<T5, T6>, o6: PipedOperator<T6, T7>, o7: PipedOperator<T7, T8>, o8: PipedOperator<T8, T9>, o9: PipedOperator<T9, T10>, o10: PipedOperator<T10, T11>, o11: PipedOperator<T11, T12>, o12: PipedOperator<T12, T13>, o13: PipedOperator<T13, T14>, o14: PipedOperator<T14, T15>, o15: PipedOperator<T15, T16>, o16: PipedOperator<T16, T17>, o17: PipedOperator<T17, T18>, o18: PipedOperator<T18, O>): IStreamBuilder<O>
+  // prettier-ignore
+  pipe<T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, O>(o1: PipedOperator<T, T2>, o2: PipedOperator<T2, T3>, o3: PipedOperator<T3, T4>, o4: PipedOperator<T4, T5>, o5: PipedOperator<T5, T6>, o6: PipedOperator<T6, T7>, o7: PipedOperator<T7, T8>, o8: PipedOperator<T8, T9>, o9: PipedOperator<T9, T10>, o10: PipedOperator<T10, T11>, o11: PipedOperator<T11, T12>, o12: PipedOperator<T12, T13>, o13: PipedOperator<T13, T14>, o14: PipedOperator<T14, T15>, o15: PipedOperator<T15, T16>, o16: PipedOperator<T16, T17>, o17: PipedOperator<T17, T18>, o18: PipedOperator<T18, T19>, o19: PipedOperator<T19, O>): IStreamBuilder<O>
+  // prettier-ignore
+  pipe<T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, O>(o1: PipedOperator<T, T2>, o2: PipedOperator<T2, T3>, o3: PipedOperator<T3, T4>, o4: PipedOperator<T4, T5>, o5: PipedOperator<T5, T6>, o6: PipedOperator<T6, T7>, o7: PipedOperator<T7, T8>, o8: PipedOperator<T8, T9>, o9: PipedOperator<T9, T10>, o10: PipedOperator<T10, T11>, o11: PipedOperator<T11, T12>, o12: PipedOperator<T12, T13>, o13: PipedOperator<T13, T14>, o14: PipedOperator<T14, T15>, o15: PipedOperator<T15, T16>, o16: PipedOperator<T16, T17>, o17: PipedOperator<T17, T18>, o18: PipedOperator<T18, T19>, o19: PipedOperator<T19, T20>, o20: PipedOperator<T20, O>): IStreamBuilder<O>
+
+  pipe(...operators: Array<PipedOperator<any, any>>): IStreamBuilder<any> {
+    return operators.reduce((stream, operator) => {
+      return operator(stream)
+    }, this as IStreamBuilder<any>)
+  }
+}
+
+export class RootStreamBuilder<T> extends StreamBuilder<T> {
+  sendData(collection: MultiSet<T> | MultiSetArray<T>): void {
+    this.writer.sendData(collection)
+  }
+}