fix(zqlite): convert sqlite types before pushing to outputs (rocicorp#2291)

Author: darkgnotic

packages/zero-cache/src/services/view-syncer/pipeline-driver.ts

@@ -1,8 +1,6 @@
 import {LogContext} from '@rocicorp/logger';
 import {TableSource} from '@rocicorp/zqlite/src/table-source.js';
 import {assert} from 'shared/src/asserts.js';
-import {must} from 'shared/src/must.js';
-import {JSONValue} from 'zero-cache/src/types/bigint-json.js';
 import {mapLiteDataTypeToZqlValueType} from 'zero-cache/src/types/lite.js';
 import {AST} from 'zql/src/zql/ast/ast.js';
 import {buildPipeline} from 'zql/src/zql/builder/builder.js';
@@ -190,28 +188,13 @@ export class PipelineDriver {
     };
   }
 
-  #convertTypes(table: string, row: Record<string, JSONValue>): Row {
-    assert(this.#tableSpecs);
-    const spec = must(this.#tableSpecs.get(table));
-    for (const col of spec.boolColumns) {
-      row[col] = !!row[col];
-    }
-    return row as Row;
-  }
-
   *#advance(diff: SnapshotDiff): Iterable<RowChange> {
     for (const {table, prevValue, nextValue} of diff) {
       if (prevValue) {
-        yield* this.#push(table, {
-          type: 'remove',
-          row: this.#convertTypes(table, prevValue),
-        });
+        yield* this.#push(table, {type: 'remove', row: prevValue as Row});
       }
       if (nextValue) {
-        yield* this.#push(table, {
-          type: 'add',
-          row: this.#convertTypes(table, nextValue),
-        });
+        yield* this.#push(table, {type: 'add', row: nextValue as Row});
       }
     }
 

packages/zqlite/src/table-source.test.ts

@@ -1,6 +1,7 @@
 import Database from 'better-sqlite3';
 import {describe, expect, test} from 'vitest';
 import {Catch} from 'zql/src/zql/ivm/catch.js';
+import {Change} from 'zql/src/zql/ivm/change.js';
 import {makeComparator} from 'zql/src/zql/ivm/data.js';
 import {ValueType} from 'zql/src/zql/ivm/schema.js';
 import {runCases} from 'zql/src/zql/ivm/test/source-cases.js';
@@ -196,17 +197,26 @@ describe('fetching from a table source', () => {
   });
 });
 
-test('pushing values does the correct writes', () => {
+test('pushing values does the correct writes and outputs', () => {
   const db1 = new Database(':memory:');
   const db2 = new Database(':memory:');
   db1.exec(/* sql */ `CREATE TABLE foo (a, b, c, PRIMARY KEY (a, b));`);
   db2.exec(/* sql */ `CREATE TABLE foo (a, b, c, PRIMARY KEY (a, b));`);
   const source = new TableSource(
     db1,
     'foo',
-    {a: 'number', b: 'number', c: 'number'},
+    {a: 'number', b: 'number', c: 'boolean'},
     ['a', 'b'],
   );
+  const outputted: Change[] = [];
+  source
+    .connect([
+      ['a', 'asc'],
+      ['b', 'asc'],
+    ])
+    .setOutput({
+      push: change => outputted.push(change),
+    });
 
   for (const db of [db1, db2]) {
     const read = db.prepare('SELECT * FROM foo');
@@ -221,16 +231,37 @@ test('pushing values does the correct writes', () => {
      */
     source.push({
       type: 'add',
-      row: {a: 1, b: 2, c: 3},
+      row: {a: 1, b: 2, c: 0},
     });
 
-    expect(read.all()).toEqual([{a: 1, b: 2, c: 3}]);
+    expect(outputted.shift()).toEqual({
+      type: 'add',
+      node: {
+        relationships: {},
+        row: {
+          a: 1,
+          b: 2,
+          c: false,
+        },
+      },
+    });
+    expect(read.all()).toEqual([{a: 1, b: 2, c: 0}]);
 
     source.push({
       type: 'remove',
       row: {a: 1, b: 2},
     });
 
+    expect(outputted.shift()).toEqual({
+      type: 'remove',
+      node: {
+        relationships: {},
+        row: {
+          a: 1,
+          b: 2,
+        },
+      },
+    });
     expect(read.all()).toEqual([]);
 
     expect(() => {
@@ -240,10 +271,25 @@ test('pushing values does the correct writes', () => {
       });
     }).toThrow();
 
+    expect(read.all()).toEqual([]);
+
     source.push({
       type: 'add',
-      row: {a: 1, b: 2, c: 3},
+      row: {a: 1, b: 2, c: 1},
+    });
+
+    expect(outputted.shift()).toEqual({
+      type: 'add',
+      node: {
+        relationships: {},
+        row: {
+          a: 1,
+          b: 2,
+          c: true,
+        },
+      },
     });
+    expect(read.all()).toEqual([{a: 1, b: 2, c: 1}]);
 
     expect(() => {
       source.push({

packages/zqlite/src/table-source.ts

@@ -288,6 +288,8 @@ export class TableSource implements Source {
       assert(exists, 'Row not found');
     }
 
+    // Outputs should see converted types (e.g. boolean).
+    fromSQLiteTypes(this.#columns, change.row);
     for (const [outputIndex, {output}] of this.#connections.entries()) {
       this.#overlay = {outputIndex, change};
       if (output) {