Support raw tables

Author: simolus3

.changeset/bright-snakes-clean.md

@@ -0,0 +1,37 @@
+---
+'@powersync/common': minor
+'@powersync/node': minor
+'@powersync/web': minor
+'@powersync/react-native': minor
+---
+
+Add experimental support for raw tables, giving you full control over the table structure to sync into.
+While PowerSync manages tables as JSON views by default, raw tables have to be created by the application
+developer. Also, the upsert and delete statements for raw tables needs to be specified in the app schema:
+
+```JavaScript
+const customSchema = new Schema({});
+customSchema.withRawTables({
+  lists: {
+    put: {
+      sql: 'INSERT OR REPLACE INTO lists (id, name) VALUES (?, ?)',
+      // put statements can use `Id` and extracted columns to bind parameters.
+      params: ['Id', { Column: 'name' }]
+    },
+    delete: {
+      sql: 'DELETE FROM lists WHERE id = ?',
+      // delete statements can only use the id (but a CTE querying existing rows by id could
+      // be used as a workaround).
+      params: ['Id']
+    }
+  }
+});
+
+const powersync = // open powersync database;
+await powersync.execute('CREATE TABLE lists (id TEXT NOT NULL PRIMARY KEY, name TEXT);');
+
+// Ready to sync into your custom table at this point
+```
+
+The main benefit of raw tables is better query performance (since SQLite doesn't have to
+extract rows from JSON) and more control (allowing the use of e.g. column and table constraints).

packages/common/rollup.config.mjs

@@ -36,8 +36,8 @@ export default (commandLineArgs) => {
         ReadableStream: ['web-streams-polyfill/ponyfill', 'ReadableStream'],
         // Used by can-ndjson-stream
         TextDecoder: ['text-encoding', 'TextDecoder']
-      }),
-      terser({ sourceMap })
+      })
+      //terser({ sourceMap })
     ],
     // This makes life easier
     external: [

packages/node/tests/sync.test.ts

@@ -7,6 +7,7 @@ import {
   OplogEntryJSON,
   PowerSyncConnectionOptions,
   ProgressWithOperations,
+  Schema,
   SyncClientImplementation,
   SyncStreamConnectionMethod
 } from '@powersync/common';
@@ -638,6 +639,85 @@ function defineSyncTests(impl: SyncClientImplementation) {
     expect(another.currentStatus.statusForPriority(0).hasSynced).toBeTruthy();
     await another.waitForFirstSync({ priority: 0 });
   });
+
+  if (impl == SyncClientImplementation.RUST) {
+    mockSyncServiceTest('raw tables', async ({ syncService }) => {
+      const customSchema = new Schema({});
+      customSchema.withRawTables({
+        lists: {
+          put: {
+            sql: 'INSERT OR REPLACE INTO lists (id, name) VALUES (?, ?)',
+            params: ['Id', { Column: 'name' }]
+          },
+          delete: {
+            sql: 'DELETE FROM lists WHERE id = ?',
+            params: ['Id']
+          }
+        }
+      });
+
+      const powersync = await syncService.createDatabase({ schema: customSchema });
+      await powersync.execute('CREATE TABLE lists (id TEXT NOT NULL PRIMARY KEY, name TEXT);');
+
+      const query = powersync.watchWithAsyncGenerator('SELECT * FROM lists')[Symbol.asyncIterator]();
+      expect((await query.next()).value.rows._array).toStrictEqual([]);
+
+      powersync.connect(new TestConnector(), options);
+      await vi.waitFor(() => expect(syncService.connectedListeners).toHaveLength(1));
+
+      syncService.pushLine({
+        checkpoint: {
+          last_op_id: '1',
+          buckets: [bucket('a', 1)]
+        }
+      });
+      syncService.pushLine({
+        data: {
+          bucket: 'a',
+          data: [
+            {
+              checksum: 0,
+              op_id: '1',
+              op: 'PUT',
+              object_id: 'my_list',
+              object_type: 'lists',
+              data: '{"name": "custom list"}'
+            }
+          ]
+        }
+      });
+      syncService.pushLine({ checkpoint_complete: { last_op_id: '1' } });
+      await powersync.waitForFirstSync();
+
+      expect((await query.next()).value.rows._array).toStrictEqual([{ id: 'my_list', name: 'custom list' }]);
+
+      syncService.pushLine({
+        checkpoint: {
+          last_op_id: '2',
+          buckets: [bucket('a', 2)]
+        }
+      });
+      await vi.waitFor(() => powersync.currentStatus.dataFlowStatus.downloading == true);
+      syncService.pushLine({
+        data: {
+          bucket: 'a',
+          data: [
+            {
+              checksum: 0,
+              op_id: '2',
+              op: 'REMOVE',
+              object_id: 'my_list',
+              object_type: 'lists'
+            }
+          ]
+        }
+      });
+      syncService.pushLine({ checkpoint_complete: { last_op_id: '2' } });
+      await vi.waitFor(() => powersync.currentStatus.dataFlowStatus.downloading == false);
+
+      expect((await query.next()).value.rows._array).toStrictEqual([]);
+    });
+  }
 }
 
 function bucket(name: string, count: number, options: { priority: number } = { priority: 3 }): BucketChecksum {

packages/node/tests/utils.ts

@@ -9,7 +9,6 @@ import {
   column,
   NodePowerSyncDatabaseOptions,
   PowerSyncBackendConnector,
-  PowerSyncConnectionOptions,
   PowerSyncCredentials,
   PowerSyncDatabase,
   Schema,
@@ -56,12 +55,12 @@ async function createDatabase(
   options: Partial<NodePowerSyncDatabaseOptions> = {}
 ): Promise<PowerSyncDatabase> {
   const database = new PowerSyncDatabase({
-    ...options,
     schema: AppSchema,
     database: {
       dbFilename: 'test.db',
       dbLocation: tmpdir
-    }
+    },
+    ...options
   });
   await database.init();
   return database;
@@ -128,8 +127,9 @@ export const mockSyncServiceTest = tempDirectoryTest.extend<{
       }
     };
 
-    const newConnection = async () => {
+    const newConnection = async (options?: Partial<NodePowerSyncDatabaseOptions>) => {
       const db = await createDatabase(tmpdir, {
+        ...options,
         remoteOptions: {
           fetchImplementation: inMemoryFetch
         }
@@ -156,7 +156,7 @@ export const mockSyncServiceTest = tempDirectoryTest.extend<{
 export interface MockSyncService {
   pushLine: (line: StreamingSyncLine) => void;
   connectedListeners: any[];
-  createDatabase: () => Promise<PowerSyncDatabase>;
+  createDatabase: (options?: Partial<NodePowerSyncDatabaseOptions>) => Promise<PowerSyncDatabase>;
 }
 
 export class TestConnector implements PowerSyncBackendConnector {

packages/web/src/worker/sync/SharedSyncImplementation.ts

@@ -239,7 +239,7 @@ export class SharedSyncImplementation
    */
   async connect(options?: PowerSyncConnectionOptions) {
     this.lastConnectOptions = options;
-    return this.connectionManager.connect(CONNECTOR_PLACEHOLDER, options);
+    return this.connectionManager.connect(CONNECTOR_PLACEHOLDER, options ?? {});
   }
 
   async disconnect() {
@@ -318,7 +318,7 @@ export class SharedSyncImplementation
       this.dbAdapter = null;
 
       if (shouldReconnect) {
-        await this.connectionManager.connect(CONNECTOR_PLACEHOLDER, this.lastConnectOptions);
+        await this.connectionManager.connect(CONNECTOR_PLACEHOLDER, this.lastConnectOptions ?? {});
       }
     }