complete stubbed tests

package.json

@@ -32,6 +32,7 @@
     "@web/dev-server-import-maps": "^0.2.0",
     "@web/test-runner": "^0.18.0",
     "@web/test-runner-playwright": "^0.11.0",
+    "fast-check": "^3.16.0",
     "nanoid": "^5.0.4",
     "playwright": "^1.41.1",
     "replicache": "14.1.0",

src/zql/ast-to-ivm/pipelineBuilder.ts

@@ -36,7 +36,7 @@ export function buildPipeline(
   return ret;
 }
 
-function applySelect(
+export function applySelect(
   stream: DifferenceStream<unknown>,
   select: string[],
   orderBy: [string[], 'asc' | 'desc'] | undefined,

src/zql/ivm/Materialite.ts

@@ -1,4 +1,6 @@
+import {Comparator} from '@vlcn.io/ds-and-algos/types';
 import {ISourceInternal} from './source/ISource.js';
+import {MutableSetSource} from './source/SetSource.js';
 import {StatelessSource} from './source/StatelessSource.js';
 import {Version} from './types.js';
 
@@ -35,6 +37,10 @@ export class Materialite {
     return new StatelessSource<T>(this.#internal);
   }
 
+  newSetSource<T>(comparator: Comparator<T>) {
+    return new MutableSetSource<T>(this.#internal, comparator);
+  }
+
   /**
    * Run the provided lambda in a transaciton.
    * Will be committed when the lambda exits

src/zql/ivm/graph/DifferenceStream.test.ts

@@ -115,40 +115,3 @@ test('map, filter, linearCount', () => {
 
   expect(expectRan).toBe(2);
 });
-
-test('the reactive graph is run in a breadth first manner', () => {
-  // we must do breadth first so each layer has prepared its outputs before the next layer
-  // tries to read those outputs.
-  expect(true).toBe(false);
-});
-
-test('operators cannot be double-notified', () => {
-  expect(true).toBe(false);
-});
-
-test(
-  'if an operator stops the computation then downstreams of that operator are not notified of commit ' +
-    '(because nothing observable changed downstream)',
-  () => {
-    expect(true).toBe(false);
-  },
-);
-
-test('binary operators successfully pull from both inputs', () => {
-  // The interesting case:
-  /*
-           s1    s2
-            \   |
-            o1  o3
-             | /
-             o2
-
-    - s1 and s2 will have queued inputs
-    - o1 will be told to run
-    - o3 should also be told to run (breadth first)
-    - o1 notifies o2
-    - o2 pulls from o3
-    - o3 should have data available (if it computed something) or null if nothing.
-  */
-  expect(true).toBe(false);
-});

src/zql/ivm/graph/DifferenceStream.ts

@@ -4,7 +4,6 @@ import {DifferenceStreamWriter} from './DifferenceStreamWriter.js';
 import {IDifferenceStream} from './IDifferenceStream.js';
 import {LinearCountOperator} from './operators/CountOperator.js';
 import {DifferenceEffectOperator} from './operators/DifferenceEffectOperator.js';
-import {EffectOperator} from './operators/EffectOperator.js';
 import {FilterOperator} from './operators/FilterOperator.js';
 import {MapOperator} from './operators/MapOperator.js';
 
@@ -50,16 +49,6 @@ export class DifferenceStream<T> implements IDifferenceStream<T> {
     return ret;
   }
 
-  /**
-   * Run some sort of side-effect against values in the stream.
-   * e.g., I/O & logging
-   */
-  effect(f: (i: T) => void) {
-    const ret = this.newStream<T>();
-    new EffectOperator(this.#writer.newReader(), ret.#writer, f);
-    return ret;
-  }
-
   /**
    * Runs a side-effect for all events in the stream.
    * If `mult < 0` that means the value V was retracted `mult` times.
@@ -68,7 +57,7 @@ export class DifferenceStream<T> implements IDifferenceStream<T> {
    * @param f
    * @returns
    */
-  differenceEffect(f: (i: T, mult: number) => void) {
+  effect(f: (i: T, mult: number) => void) {
     const ret = this.newStream<T>();
     new DifferenceEffectOperator(this.#writer.newReader(), ret.#writer, f);
     return ret;

src/zql/ivm/graph/IDifferenceStream.ts

@@ -23,5 +23,5 @@ export interface IDifferenceStream<T> {
   filter<S extends T>(f: (x: T) => x is S): IDifferenceStream<S>;
   filter(f: (x: T) => boolean): IDifferenceStream<T>;
   linearCount(): IDifferenceStream<number>;
-  effect(f: (i: T) => void): IDifferenceStream<T>;
+  effect(f: (i: T, m: number) => void): IDifferenceStream<T>;
 }

src/zql/ivm/graph/operators/DifferenceEffectOperator.test.ts

@@ -0,0 +1,44 @@
+import {expect, test} from 'vitest';
+import {DifferenceStreamWriter} from '../DifferenceStreamWriter.js';
+import {DifferenceEffectOperator} from './DifferenceEffectOperator.js';
+import {Multiset} from '../../Multiset.js';
+
+test('calls effect with raw difference events', () => {
+  const inputWriter = new DifferenceStreamWriter<number>();
+  const inputReader = inputWriter.newReader();
+  const output = new DifferenceStreamWriter<number>();
+
+  let called = false;
+  let value = 0;
+  let mult = 0;
+  new DifferenceEffectOperator(inputReader, output, (v, m) => {
+    called = true;
+    value = v;
+    mult = m;
+  });
+
+  inputWriter.queueData([1, new Multiset([[1, 1]])]);
+  inputWriter.notify(1);
+
+  // effect not run until commit
+  expect(called).toBe(false);
+
+  inputWriter.notifyCommitted(1);
+  expect(called).toBe(true);
+  expect(value).toBe(1);
+  expect(mult).toBe(1);
+
+  called = false;
+  value = 0;
+  mult = 0;
+  inputWriter.queueData([2, new Multiset([[1, -1]])]);
+  inputWriter.notify(2);
+
+  // effect not run until commit
+  expect(called).toBe(false);
+
+  inputWriter.notifyCommitted(2);
+  expect(called).toBe(true);
+  expect(value).toBe(1);
+  expect(mult).toBe(-1);
+});

src/zql/ivm/graph/operators/FilterOperator.test.ts

@@ -0,0 +1,100 @@
+import {expect, test} from 'vitest';
+import {FilterOperator} from './FilterOperator.js';
+import {DifferenceStreamWriter} from '../DifferenceStreamWriter.js';
+import {Multiset} from '../../Multiset.js';
+import {NoOp} from './Operator.js';
+
+test('does not emit any rows that fail the filter', () => {
+  const inputWriter = new DifferenceStreamWriter<number>();
+  const inputReader = inputWriter.newReader();
+  const output = new DifferenceStreamWriter<number>();
+
+  new FilterOperator(inputReader, output, _ => false);
+
+  const outReader = output.newReader();
+  outReader.setOperator(new NoOp());
+
+  inputWriter.queueData([
+    1,
+    new Multiset([
+      [1, 1],
+      [2, 2],
+      [1, -1],
+      [2, -2],
+    ]),
+  ]);
+  inputWriter.notify(1);
+  inputWriter.notifyCommitted(1);
+  const items = outReader.drain(1);
+
+  expect(items.length).toBe(1);
+  expect([...items[0].entries].length).toBe(0);
+});
+
+test('emits all rows that pass the filter (including deletes / retractions)', () => {
+  const inputWriter = new DifferenceStreamWriter<number>();
+  const inputReader = inputWriter.newReader();
+  const output = new DifferenceStreamWriter<number>();
+
+  new FilterOperator(inputReader, output, _ => true);
+
+  const outReader = output.newReader();
+  outReader.setOperator(new NoOp());
+
+  inputWriter.queueData([
+    1,
+    new Multiset([
+      [1, 1],
+      [2, 2],
+      [1, -1],
+      [2, -2],
+    ]),
+  ]);
+  inputWriter.notify(1);
+  inputWriter.notifyCommitted(1);
+  const items = outReader.drain(1);
+
+  expect(items.length).toBe(1);
+  const entries = [...items[0].entries];
+  expect(entries).toMatchObject([
+    [1, 1],
+    [2, 2],
+    [1, -1],
+    [2, -2],
+  ]);
+});
+
+test('test that filter is lazy / the filter is not actually run until we pull on it', () => {
+  const inputWriter = new DifferenceStreamWriter<number>();
+  const inputReader = inputWriter.newReader();
+  const output = new DifferenceStreamWriter<number>();
+
+  let called = false;
+  new FilterOperator(inputReader, output, _ => {
+    called = true;
+    return true;
+  });
+
+  const outReader = output.newReader();
+  outReader.setOperator(new NoOp());
+
+  inputWriter.queueData([
+    1,
+    new Multiset([
+      [1, 1],
+      [2, 2],
+      [1, -1],
+      [2, -2],
+    ]),
+  ]);
+  inputWriter.notify(1);
+  inputWriter.notifyCommitted(1);
+
+  // we run the graph but the filter is not run until we pull on it
+  expect(called).toBe(false);
+
+  const items = outReader.drain(1);
+  // consume all the rows
+  [...items[0].entries];
+  expect(called).toBe(true);
+});