d2mini - d2ts without the multi-dimensional versioning (#65)

* clone d2ts to d2mini with a few omissions

* Stateless operators done

* version-index done

* join and consolidate done

* reduce

* count and distinct

* topK

* graph test

* orderBy tests

* filterBy

* groupby

* rename test files

* rename some stuff

* remove itterate from multiset

* use hash

* format

* tidy

* fix tests

* update package.json

* changeset

* improvments to index

* refactor index

* hash return string - forward compat with 128bit hash

* remove unused code

Author: samwillis

.changeset/silly-ads-scream.md

@@ -0,0 +1,5 @@
+---
+'@electric-sql/d2mini': patch
+---
+
+First release of D2mini - a minimal implementation of the D2TS Differential Dataflow library but simplified and without the complexities of multi-dimensional versioning.

packages/d2mini/README.md

@@ -0,0 +1,56 @@
+# D2Mini is a minimal IVM implementation based on Differential Dataflow
+
+D2Mini is a minimal implementation of the D2TS dataflow graph library but simplified and without the complexities of multi-dimensional versioning.
+
+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, v } from '@electric-sql/d2ts'
+
+// 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/d2mini/eslint.config.mjs

@@ -0,0 +1,18 @@
+import baseConfig from '../../eslint.base.mjs';
+
+export default [
+  ...baseConfig,
+  {
+    files: ['**/*.ts'],
+    ignores: ['**/dist/**', '**/node_modules/**'],
+    rules: {
+      // Package-specific rules can go here
+    },
+  },
+  {
+    files: ['**/tests/**/*.ts'],
+    rules: {
+      '@typescript-eslint/no-unused-vars': 'off',
+    },
+  },
+]; 

packages/d2mini/package.json

@@ -0,0 +1,55 @@
+{
+  "name": "@electric-sql/d2mini",
+  "version": "0.1.0",
+  "license": "Apache-2.0",
+  "description": "D2Mini is a minimal implementation of Differential Dataflow for performing in-memory incremental view maintenance.",
+  "author": "Electric DB Limited",
+  "homepage": "https://github.com/electric-sql/d2ts",
+  "repository": {
+    "type": "git",
+    "url": "https://github.com/electric-sql/d2ts.git",
+    "directory": "packages/d2mini"
+  },
+  "keywords": [
+    "differential dataflow",
+    "differential",
+    "dataflow",
+    "stream processing",
+    "sync"
+  ],
+  "type": "module",
+  "main": "dist/index.js",
+  "types": "dist/index.d.ts",
+  "exports": {
+    ".": {
+      "types": "./dist/index.d.ts",
+      "default": "./dist/index.js"
+    }
+  },
+  "scripts": {
+    "build": "tsc",
+    "test": "vitest run",
+    "lint": "eslint './src/**/*.ts' './tests/**/*.ts' --config eslint.config.mjs",
+    "lint:fix": "eslint src/**/*.ts tests/**/*.ts --fix",
+    "typecheck": "tsc --noEmit",
+    "format": "prettier --write \"src/**/*.{ts,tsx,js,jsx,json}\" \"tests/**/*.{ts,tsx,js,jsx,json}\"",
+    "format:check": "prettier --check \"src/**/*.{ts,tsx,js,jsx,json}\" \"tests/**/*.{ts,tsx,js,jsx,json}\""
+  },
+  "devDependencies": {
+    "@types/murmurhash-js": "^1.0.6",
+    "@types/node": "^22.15.3",
+    "@typescript-eslint/eslint-plugin": "^8.31.1",
+    "@typescript-eslint/parser": "^8.31.1",
+    "eslint": "^9.25.1",
+    "eslint-config-prettier": "^10.1.2",
+    "eslint-plugin-prettier": "^5.2.6",
+    "prettier": "^3.5.3",
+    "tsx": "^4.19.4",
+    "typescript": "^5.8.3",
+    "vitest": "^3.1.2"
+  },
+  "dependencies": {
+    "fractional-indexing": "^3.2.0",
+    "murmurhash-js": "^1.0.0"
+  }
+}

packages/d2mini/src/d2.ts

@@ -0,0 +1,150 @@
+import {
+  BinaryOperator,
+  DifferenceStreamWriter,
+  UnaryOperator,
+} from './graph.js'
+import { DifferenceStreamReader } from './graph.js'
+import { MultiSetArray, MultiSet } from './multiset.js'
+import { PipedOperator, IStreamBuilder, ID2 } from './types.js'
+
+export class D2 implements ID2 {
+  #streams: DifferenceStreamReader<any>[] = []
+  #operators: (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: 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)
+  }
+}