don't destructure in hot path, check multiset contents, remove extran…

…eous code

src/zql/ivm/graph/difference-stream.test.ts

@@ -194,7 +194,6 @@ test('replying to a message only notifies along the requesting path', () => {
   const s3 = stream.debug(() => notified.push(3));
 
   s1.debug(() => notified.push(4));
-  // s2.debug
   const x = new DifferenceStream();
   const s2Dbg = new DebugOperator(s2, x, () => notified.push(5));
   x.setUpstream(s2Dbg);

src/zql/ivm/graph/difference-stream.ts

@@ -49,8 +49,17 @@ export type Listener<T> = {
  */
 // T extends object: I believe in the context of ZQL we only deal with object.
 export class DifferenceStream<T extends object> {
+  /**
+   * Operators that are listening to this stream.
+   */
   readonly #downstreams: Set<Listener<T>> = new Set();
+  /**
+   * The operator that is sending data to this stream.
+   */
   #upstream: Operator | undefined;
+  /**
+   * Downstreams that requested historical data.
+   */
   readonly #requestors = new Set<Listener<T>>();
 
   addDownstream(listener: Listener<T>) {
@@ -82,11 +91,13 @@ export class DifferenceStream<T extends object> {
 
   commit(version: Version) {
     if (this.#requestors.size > 0) {
-      this.#requestors;
       for (const requestor of this.#requestors) {
         try {
           requestor.commit(version);
         } catch (e) {
+          // `commit` notifies client code
+          // If client code throws we'll put IVM back into a consistent state
+          // by clearing the requestors.
           this.#requestors.clear();
           throw e;
         }

src/zql/ivm/graph/operators/filter-operator.test.ts

@@ -75,7 +75,13 @@ test('test that filter is lazy / the filter is not actually run until we pull on
 
   // consume all the rows
   for (const m of msgs) {
-    [...m];
+    expect([...m]).toEqual([
+      [{id: 1}, 1],
+      [{id: 2}, 2],
+      [{id: 1}, -1],
+      [{id: 2}, -2],
+    ]);
   }
+
   expect(called).toBe(true);
 });

src/zql/ivm/graph/operators/filter-operator.ts

@@ -12,8 +12,6 @@ export class FilterOperator<I extends object> extends LinearUnaryOperator<
     output: DifferenceStream<I>,
     f: (input: I) => boolean,
   ) {
-    super(input, output, (data: Multiset<I>) =>
-      genFilter(data, ([value, _]) => f(value)),
-    );
+    super(input, output, (data: Multiset<I>) => genFilter(data, e => f(e[0])));
   }
 }

src/zql/ivm/graph/operators/operator.ts

@@ -16,7 +16,7 @@ export class NoOp implements Operator {
   constructor() {}
   commit(_v: Version): void {}
   messageUpstream(): void {}
-  destroy() {}
+  destroy(): void {}
 }
 
 /**