feat(zql): Add and/or to query API

This commit adds the ability to use `and` and `or` in the query API.
Typical usage would be:

```ts
q.where(
  or(
    and(
      expression('owner', '=', 'arv'),
      expression('priority', '=', 'high'),
    ),
    expression('status', '=', 'open'),
  ),
)
```

I think we want to design the surface API further but this unblocks us.

This commit also contains a few fixes:

- MutableTreeView was incorrectly telling the AbstractView not to call
  the listeners when it got an update that didn't change the view. But,
  with branches in the graph we can have multiple calls to `_newData`.
- With branches in the graph we the Source can get
  the same `PullMsg` more than once. Switch to a set to avoid this.

src/zql/ast-to-ivm/pipeline-builder.test.ts

@@ -143,14 +143,26 @@ describe('OR', () => {
     b: number;
   };
 
+  type DeleteE = {
+    delete: E;
+  };
+
   type NoUndefined<T> = T extends undefined ? never : T;
 
   type Case = {
+    name?: string | undefined;
     where: NoUndefined<AST['where']>;
-    values: E[];
-    expected: string[];
+    values?: (E | DeleteE)[] | undefined;
+    expected: (E | [v: E, multiplicity: number])[];
   };
 
+  const defaultValues: (E | DeleteE)[] = [
+    {id: 'a', a: 1, b: 1},
+    {id: 'b', a: 2, b: 1},
+    {id: 'c', a: 1, b: 2},
+    {id: 'd', a: 2, b: 2},
+  ];
+
   const cases: Case[] = [
     {
       where: {
@@ -160,26 +172,21 @@ describe('OR', () => {
           {op: '=', field: 'b', value: {type: 'literal', value: 2}},
         ],
       },
-      values: [
+      expected: [
         {id: 'a', a: 1, b: 1},
-        {id: 'b', a: 2, b: 1},
         {id: 'c', a: 1, b: 2},
         {id: 'd', a: 2, b: 2},
       ],
-      expected: ['a', 'c', 'd'],
     },
     {
       where: {
         op: 'OR',
         conditions: [{op: '=', field: 'a', value: {type: 'literal', value: 1}}],
       },
-      values: [
+      expected: [
         {id: 'a', a: 1, b: 1},
-        {id: 'b', a: 2, b: 1},
         {id: 'c', a: 1, b: 2},
-        {id: 'd', a: 2, b: 2},
       ],
-      expected: ['a', 'c'],
     },
     {
       where: {
@@ -196,7 +203,11 @@ describe('OR', () => {
         {id: 'c', a: 1, b: 2},
         {id: 'd', a: 3, b: 3},
       ],
-      expected: ['a', 'b', 'c'],
+      expected: [
+        {id: 'a', a: 1, b: 1},
+        {id: 'b', a: 2, b: 1},
+        {id: 'c', a: 1, b: 2},
+      ],
     },
     {
       where: {
@@ -206,13 +217,10 @@ describe('OR', () => {
           {op: '=', field: 'a', value: {type: 'literal', value: 1}},
         ],
       },
-      values: [
+      expected: [
         {id: 'a', a: 1, b: 1},
-        {id: 'b', a: 2, b: 1},
         {id: 'c', a: 1, b: 2},
-        {id: 'd', a: 2, b: 2},
       ],
-      expected: ['a', 'c'],
     },
     {
       where: {
@@ -234,13 +242,10 @@ describe('OR', () => {
           },
         ],
       },
-      values: [
+      expected: [
         {id: 'a', a: 1, b: 1},
-        {id: 'b', a: 2, b: 1},
-        {id: 'c', a: 1, b: 2},
         {id: 'd', a: 2, b: 2},
       ],
-      expected: ['a', 'd'],
     },
 
     {
@@ -263,13 +268,10 @@ describe('OR', () => {
           },
         ],
       },
-      values: [
-        {id: 'a', a: 1, b: 1},
+      expected: [
         {id: 'b', a: 2, b: 1},
         {id: 'c', a: 1, b: 2},
-        {id: 'd', a: 2, b: 2},
       ],
-      expected: ['b', 'c'],
     },
 
     {
@@ -292,16 +294,15 @@ describe('OR', () => {
           },
         ],
       },
-      values: [
+      expected: [
         {id: 'a', a: 1, b: 1},
         {id: 'b', a: 2, b: 1},
         {id: 'c', a: 1, b: 2},
-        {id: 'd', a: 2, b: 2},
       ],
-      expected: ['a', 'b', 'c'],
     },
 
     {
+      name: 'Repeat identical conditions',
       where: {
         op: 'OR',
         conditions: [
@@ -310,13 +311,10 @@ describe('OR', () => {
           {op: '=', field: 'a', value: {type: 'literal', value: 1}},
         ],
       },
-      values: [
+      expected: [
         {id: 'a', a: 1, b: 1},
-        {id: 'b', a: 2, b: 1},
         {id: 'c', a: 1, b: 2},
-        {id: 'd', a: 2, b: 2},
       ],
-      expected: ['a', 'c'],
     },
 
     {
@@ -328,27 +326,76 @@ describe('OR', () => {
           {op: '=', field: 'a', value: {type: 'literal', value: 5}},
         ],
       },
+      expected: [],
+    },
+
+    {
+      where: {
+        op: 'AND',
+        conditions: [
+          {
+            op: 'OR',
+            conditions: [
+              {op: '=', field: 'a', value: {type: 'literal', value: 1}},
+              {op: '=', field: 'a', value: {type: 'literal', value: 2}},
+              {op: '=', field: 'a', value: {type: 'literal', value: 3}},
+            ],
+          },
+          {
+            op: '=',
+            field: 'b',
+            value: {type: 'literal', value: 1},
+          },
+        ],
+      },
       values: [
         {id: 'a', a: 1, b: 1},
-        {id: 'b', a: 2, b: 1},
-        {id: 'c', a: 1, b: 2},
-        {id: 'd', a: 2, b: 2},
+        {id: 'b', a: 2, b: 2},
+        {id: 'c', a: 3, b: 1},
+        {id: 'd', a: 4, b: 1},
+      ],
+      expected: [
+        {id: 'a', a: 1, b: 1},
+        {id: 'c', a: 3, b: 1},
+      ],
+    },
+
+    {
+      name: 'With delete',
+      where: {
+        op: 'OR',
+        conditions: [
+          {op: '=', field: 'a', value: {type: 'literal', value: 1}},
+          {op: '=', field: 'a', value: {type: 'literal', value: 2}},
+        ],
+      },
+      values: [
+        {id: 'a', a: 1, b: 1},
+        // Even though it is really nonsensical to delete this entry since this
+        // entry does not exist in the model it should still work.
+        {delete: {id: 'a', a: 1, b: 3}},
+        {id: 'a', a: 2, b: 2},
+        {delete: {id: 'c', a: 3, b: 2}},
+      ],
+      expected: [
+        {id: 'a', a: 1, b: 1},
+        [{id: 'a', a: 1, b: 3}, -1],
+        {id: 'a', a: 2, b: 2},
       ],
-      expected: [],
     },
   ];
 
   const comparator = (l: E, r: E) => compareUTF8(l.id, r.id);
 
   for (const c of cases) {
-    test(conditionToString(c.where), () => {
-      const {values} = c;
+    test((c.name ? c.name + ': ' : '') + conditionToString(c.where), () => {
+      const {values = defaultValues} = c;
       const m = new Materialite();
       const s = m.newSetSource<E>(comparator);
 
       const ast: AST = {
-        table: 'e1',
-        select: ['id'],
+        table: 'items',
+        select: ['id', 'a', 'b'],
         where: c.where,
         orderBy: [['id'], 'asc'],
       };
@@ -359,12 +406,21 @@ describe('OR', () => {
       );
 
       const log: unknown[] = [];
-      pipeline.effect(x => {
-        log.push(x.id);
+      pipeline.effect((value, multiplicity) => {
+        if (multiplicity === 1) {
+          log.push(value);
+        } else {
+          log.push([value, multiplicity]);
+        }
       });
 
       for (const value of values) {
-        s.add(value);
+        if ('delete' in value) {
+          s.delete(value.delete);
+          continue;
+        } else {
+          s.add(value);
+        }
       }
 
       expect(log).toEqual(c.expected);

src/zql/ast-to-ivm/pipeline-builder.ts

@@ -8,7 +8,7 @@ import {
   SimpleOperator,
 } from '../ast/ast.js';
 import {must} from '../error/asserts.js';
-import {DifferenceStream} from '../ivm/graph/difference-stream.js';
+import {DifferenceStream, concat} from '../ivm/graph/difference-stream.js';
 
 export const orderingProp = Symbol();
 
@@ -152,8 +152,8 @@ function applyOr<T extends Entity>(
   // Or is done by branching the stream and then applying the conditions to each
   // branch. Then we merge the branches back together. At this point we need to
   // ensure we do not get duplicate entries so we add a distinct operator
-  const [first, ...rest] = conditions.map(c => applyWhere(stream, c));
-  return first.concat(...rest).distinct();
+  const branches = conditions.map(c => applyWhere(stream, c));
+  return concat(branches).distinct();
 }
 
 function applySimpleCondition<T extends Entity>(

src/zql/integration.test.ts

@@ -1,12 +1,12 @@
+import fc from 'fast-check';
+import {nanoid} from 'nanoid';
+import {Replicache, TEST_LICENSE_KEY} from 'replicache';
 import {expect, test} from 'vitest';
 import {z} from 'zod';
 import {generate} from '../generate.js';
 import {makeReplicacheContext} from './context/replicache-context.js';
-import {Replicache, TEST_LICENSE_KEY} from 'replicache';
-import {nanoid} from 'nanoid';
-import fc from 'fast-check';
-import {EntityQuery} from './query/entity-query.js';
 import * as agg from './query/agg.js';
+import {EntityQuery, expression, or} from './query/entity-query.js';
 
 export async function tickAFewTimes(n = 10, time = 0) {
   for (let i = 0; i < n; i++) {
@@ -570,3 +570,55 @@ test('write delay with 1, 10, 100, 1000s of active queries', () => {});
 test('asc/desc difference does not create new sources', () => {});
 
 test('we do not do a full scan when the source order matches the view order', () => {});
+
+test('or where', async () => {
+  const {q, r} = setup();
+  const issues: Issue[] = [
+    {
+      id: 'a',
+      title: 'foo',
+      status: 'open',
+      priority: 'high',
+      assignee: 'charles',
+      created: Date.now(),
+      updated: Date.now(),
+    },
+    {
+      id: 'b',
+      title: 'bar',
+      status: 'open',
+      priority: 'medium',
+      assignee: 'bob',
+      created: Date.now(),
+      updated: Date.now(),
+    },
+    {
+      id: 'c',
+      title: 'baz',
+      status: 'closed',
+      priority: 'low',
+      assignee: 'alice',
+      created: Date.now(),
+      updated: Date.now(),
+    },
+  ] as const;
+  await Promise.all(issues.map(r.mutate.initIssue));
+
+  const stmt = q
+    .select('id')
+    .where(
+      or(
+        expression('status', '=', 'open'),
+        expression('priority', '>=', 'medium'),
+      ),
+    )
+    .prepare();
+  const rows = await stmt.exec();
+  expect(rows).toEqual([{id: 'a'}, {id: 'b'}]);
+
+  await r.mutate.deleteIssue('a');
+  const rows2 = await stmt.exec();
+  expect(rows2).toEqual([{id: 'b'}]);
+
+  await r.close();
+});

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

@@ -127,11 +127,6 @@ export class DifferenceStream<T extends object> {
     );
   }
 
-  concat(...others: DifferenceStream<T>[]): DifferenceStream<T> {
-    const stream = new DifferenceStream<T>();
-    return stream.setUpstream(new ConcatOperator([this, ...others], stream));
-  }
-
   distinct(): DifferenceStream<T> {
     const stream = new DifferenceStream<T>();
     return stream.setUpstream(
@@ -209,3 +204,10 @@ export class DifferenceStream<T extends object> {
     }
   }
 }
+
+export function concat<T extends object>(
+  streams: DifferenceStream<T>[],
+): DifferenceStream<T> {
+  const stream = new DifferenceStream<T>();
+  return stream.setUpstream(new ConcatOperator(streams, stream));
+}