Merge remote-tracking branch 'origin/drizzle-read-only-queries' into drizzle-fix

Author: stevensJourney

.changeset/hot-socks-love.md

@@ -0,0 +1,6 @@
+---
+'@powersync/drizzle-driver': minor
+'@powersync/node': minor
+---
+
+Add support for concurrent read queries with Drizzle.

packages/drizzle-driver/package.json

@@ -50,9 +50,9 @@
     "@powersync/web": "workspace:*",
     "@journeyapps/wa-sqlite": "^1.3.2",
     "@types/node": "^20.17.6",
-    "drizzle-orm": "^0.35.2",
+    "drizzle-orm": "^0.44.7",
     "vite": "^6.1.0",
     "vite-plugin-top-level-await": "^1.4.4",
     "vite-plugin-wasm": "^3.3.0"
   }
-}
+}

packages/drizzle-driver/src/sqlite/PowerSyncSQLiteBaseSession.ts

@@ -1,4 +1,5 @@
-import { QueryResult } from '@powersync/common';
+import type { QueryResult } from '@powersync/common';
+import type { WithCacheConfig } from 'drizzle-orm/cache/core/types';
 import { entityKind } from 'drizzle-orm/entity';
 import type { Logger } from 'drizzle-orm/logger';
 import { NoopLogger } from 'drizzle-orm/logger';
@@ -53,7 +54,12 @@ export class PowerSyncSQLiteBaseSession<
     fields: SelectedFieldsOrdered | undefined,
     executeMethod: SQLiteExecuteMethod,
     isResponseInArrayMode: boolean,
-    customResultMapper?: (rows: unknown[][], mapColumnValue?: (value: unknown) => unknown) => unknown
+    customResultMapper?: (rows: unknown[][], mapColumnValue?: (value: unknown) => unknown) => unknown,
+    queryMetadata?: {
+      type: 'select' | 'update' | 'delete' | 'insert';
+      tables: string[];
+    },
+    cacheConfig?: WithCacheConfig
   ): PowerSyncSQLitePreparedQuery<T> {
     return new PowerSyncSQLitePreparedQuery(
       this.contextProvider,
@@ -62,7 +68,10 @@ export class PowerSyncSQLiteBaseSession<
       fields,
       executeMethod,
       isResponseInArrayMode,
-      customResultMapper
+      customResultMapper,
+      undefined, // cache not supported yet
+      queryMetadata,
+      cacheConfig
     );
   }
 

packages/drizzle-driver/src/sqlite/PowerSyncSQLitePreparedQuery.ts

@@ -1,5 +1,7 @@
-import { LockContext, QueryResult } from '@powersync/common';
-import { Column, DriverValueDecoder, SQL, getTableName } from 'drizzle-orm';
+import type { QueryResult } from '@powersync/common';
+import { Column, DriverValueDecoder, getTableName, SQL } from 'drizzle-orm';
+import type { Cache } from 'drizzle-orm/cache/core';
+import type { WithCacheConfig } from 'drizzle-orm/cache/core/types';
 import { entityKind, is } from 'drizzle-orm/entity';
 import type { Logger } from 'drizzle-orm/logger';
 import { fillPlaceholders, type Query } from 'drizzle-orm/sql/sql';
@@ -42,16 +44,27 @@ export class PowerSyncSQLitePreparedQuery<
 }> {
   static readonly [entityKind]: string = 'PowerSyncSQLitePreparedQuery';
 
+  private readOnly = false;
+
   constructor(
     private contextProvider: ContextProvider,
     query: Query,
     private logger: Logger,
     private fields: SelectedFieldsOrdered | undefined,
     executeMethod: SQLiteExecuteMethod,
     private _isResponseInArrayMode: boolean,
-    private customResultMapper?: (rows: unknown[][]) => unknown
+    private customResultMapper?: (rows: unknown[][]) => unknown,
+    cache?: Cache | undefined,
+    queryMetadata?:
+      | {
+          type: 'select' | 'update' | 'delete' | 'insert';
+          tables: string[];
+        }
+      | undefined,
+    cacheConfig?: WithCacheConfig | undefined
   ) {
-    super('async', executeMethod, query);
+    super('async', executeMethod, query, cache, queryMetadata, cacheConfig);
+    this.readOnly = queryMetadata?.type == 'select';
   }
 
   async run(placeholderValues?: Record<string, unknown>): Promise<QueryResult> {
@@ -115,9 +128,15 @@ export class PowerSyncSQLitePreparedQuery<
     const params = fillPlaceholders(this.query.params, placeholderValues ?? {});
     this.logger.logQuery(this.query.sql, params);
 
-    return await this.contextProvider.useReadContext(async (ctx) => {
-      return await ctx.executeRaw(this.query.sql, params);
-    });
+    if (this.readOnly) {
+      return await this.contextProvider.useReadContext(async (ctx) => {
+        return await ctx.executeRaw(this.query.sql, params);
+      });
+    } else {
+      return await this.contextProvider.useWriteContext(async (ctx) => {
+        return await ctx.executeRaw(this.query.sql, params);
+      });
+    }
   }
 
   isResponseInArrayMode(): boolean {

packages/drizzle-driver/src/sqlite/QueryContext.ts

@@ -0,0 +1,58 @@
+import { AbstractPowerSyncDatabase, LockContext, QueryResult } from '@powersync/common';
+
+/**
+ * Like LockContext, but only includes the specific methods needed for Drizzle.
+ *
+ * We extend it by adding getAllRaw, to support read-only queries with executeRaw.
+ */
+export interface QueryContext {
+  execute(query: string, params?: any[] | undefined): Promise<QueryResult>;
+  executeRaw(query: string, params?: any[] | undefined): Promise<any[][]>;
+
+  get<T>(sql: string, parameters?: any[]): Promise<T>;
+  getAll<T>(sql: string, parameters?: any[]): Promise<T[]>;
+  /**
+   * Like executeRaw, but for read-only queries.
+   */
+  getAllRaw(query: string, params?: any[] | undefined): Promise<any[][]>;
+}
+
+export class DatabaseQueryContext implements QueryContext {
+  constructor(private db: AbstractPowerSyncDatabase) {}
+  execute(query: string, params?: any[] | undefined): Promise<QueryResult> {
+    return this.db.execute(query, params);
+  }
+  executeRaw(query: string, params?: any[] | undefined) {
+    return this.db.executeRaw(query, params);
+  }
+  get<T>(sql: string, parameters?: any[]) {
+    return this.db.get<T>(sql, parameters);
+  }
+  getAll<T>(sql: string, parameters?: any[]) {
+    return this.db.getAll<T>(sql, parameters);
+  }
+  getAllRaw(query: string, params?: any[] | undefined) {
+    return this.db.readLock(async (ctx) => {
+      return ctx.executeRaw(query, params);
+    });
+  }
+}
+
+export class LockQueryContext implements QueryContext {
+  constructor(private ctx: LockContext) {}
+  execute(query: string, params?: any[] | undefined): Promise<QueryResult> {
+    return this.ctx.execute(query, params);
+  }
+  executeRaw(query: string, params?: any[] | undefined) {
+    return this.ctx.executeRaw(query, params);
+  }
+  get<T>(sql: string, parameters?: any[]) {
+    return this.ctx.get<T>(sql, parameters);
+  }
+  getAll<T>(sql: string, parameters?: any[]) {
+    return this.ctx.getAll<T>(sql, parameters);
+  }
+  getAllRaw(query: string, params?: any[] | undefined) {
+    return this.ctx.executeRaw(query, params);
+  }
+}

packages/node/package.json

@@ -76,7 +76,7 @@
     "@powersync/drizzle-driver": "workspace:*",
     "@types/node": "^24.2.0",
     "better-sqlite3": "^12.2.0",
-    "drizzle-orm": "^0.35.2",
+    "drizzle-orm": "^0.44.7",
     "rollup": "4.14.3",
     "typescript": "^5.5.3",
     "vitest": "^3.2.4"
@@ -88,4 +88,4 @@
     "real-time data stream",
     "live data"
   ]
-}
+}

packages/node/tests/DrizzleNode.test.ts

@@ -1,7 +1,7 @@
 import { sqliteTable, text } from 'drizzle-orm/sqlite-core';
 import { eq, relations } from 'drizzle-orm';
 
-import { databaseTest } from './utils';
+import { customDatabaseTest, databaseTest } from './utils';
 import { wrapPowerSyncWithDrizzle } from '@powersync/drizzle-driver';
 import { PowerSyncDatabase } from '../lib';
 import { expect } from 'vitest';
@@ -53,6 +53,15 @@ databaseTest('should retrieve a list with todos', async ({ database }) => {
   expect(result).toEqual([{ id: '1', name: 'list 1', todos: [{ id: '33', content: 'Post content', list_id: '1' }] }]);
 });
 
+databaseTest('insert returning', async ({ database }) => {
+  const db = await setupDrizzle(database);
+
+  // This is a special case since it's an insert query that returns values
+  const result = await db.insert(drizzleLists).values({ id: '2', name: 'list 2' }).returning();
+
+  expect(result).toEqual([{ id: '2', name: 'list 2' }]);
+});
+
 databaseTest('should retrieve a todo with its list', async ({ database }) => {
   const db = await setupDrizzle(database);
 
@@ -97,3 +106,126 @@ databaseTest('should return a list and todos using fullJoin', async ({ database
   expect(result[0].lists).toEqual({ id: '1', name: 'list 1' });
   expect(result[0].todos).toEqual({ id: '33', content: 'Post content', list_id: '1' });
 });
+
+customDatabaseTest({ database: { readWorkerCount: 2 } as any })(
+  'should execute transactions concurrently',
+  {},
+  async ({ database }) => {
+    // This test opens one write transaction and two read transactions, testing that they can all execute
+    // concurrently.
+    const db = await setupDrizzle(database);
+
+    const openedWrite = deferred();
+    const openedRead1 = deferred();
+    const openedRead2 = deferred();
+    const completedWrite = deferred();
+
+    const t1 = db.transaction(async (tx) => {
+      await tx.insert(drizzleLists).values({ id: '2', name: 'list 2' });
+
+      openedWrite.resolve();
+
+      await openedRead1.promise;
+      await openedRead2.promise;
+
+      completedWrite.resolve();
+    });
+
+    await openedWrite.promise;
+
+    const t2 = db.transaction(
+      async (tx) => {
+        const result = await tx.query.lists.findMany();
+        expect(result).toEqual([{ id: '1', name: 'list 1' }]);
+        openedRead1.resolve();
+        await completedWrite.promise;
+      },
+      { accessMode: 'read only' }
+    );
+
+    const t3 = db.transaction(
+      async (tx) => {
+        await openedRead1.promise;
+        const result = await tx.query.lists.findMany();
+        expect(result).toEqual([{ id: '1', name: 'list 1' }]);
+
+        openedRead2.resolve();
+        await completedWrite.promise;
+      },
+      { accessMode: 'read only' }
+    );
+
+    await Promise.all([t1, t2, t3]);
+
+    const result = await db.query.lists.findMany();
+    expect(result).toEqual([
+      { id: '1', name: 'list 1' },
+      { id: '2', name: 'list 2' }
+    ]);
+  }
+);
+
+customDatabaseTest({ database: { readWorkerCount: 2 } as any })(
+  'should execute select queries concurrently',
+  async ({ database }) => {
+    // This test opens one write transaction and two read transactions, testing that they can all execute
+    // concurrently.
+    const db = await setupDrizzle(database);
+
+    const openedWrite = deferred();
+    const completedRead = deferred();
+    const completedWrite = deferred();
+
+    const t1 = db.transaction(async (tx) => {
+      await tx.insert(drizzleLists).values({ id: '2', name: 'list 2' });
+
+      openedWrite.resolve();
+
+      await completedRead.promise;
+
+      completedWrite.resolve();
+    });
+
+    await openedWrite.promise;
+
+    const result1 = await db.select().from(drizzleLists);
+    expect(result1).toEqual([{ id: '1', name: 'list 1' }]);
+
+    const result2 = await db
+      .select()
+      .from(drizzleLists)
+      .fullJoin(drizzleTodos, eq(drizzleLists.id, drizzleTodos.list_id));
+
+    expect(result2[0].lists).toEqual({ id: '1', name: 'list 1' });
+    expect(result2[0].todos).toEqual({ id: '33', content: 'Post content', list_id: '1' });
+
+    // Note: This case is not supported yet (drizzle 0.44.7), since it doesn't set
+    // queryMetadata for these queries
+    // const result3 = await db.query.lists.findMany();
+    // expect(result3).toEqual([{ id: '1', name: 'list 1' }]);
+
+    completedRead.resolve();
+
+    await t1;
+
+    const resultAfter = await db.query.lists.findMany();
+    expect(resultAfter).toEqual([
+      { id: '1', name: 'list 1' },
+      { id: '2', name: 'list 2' }
+    ]);
+  }
+);
+
+function deferred<T = void>() {
+  let resolve: (value: T) => void;
+  let reject: (err) => void;
+  const promise = new Promise<T>((a, b) => {
+    resolve = a;
+    reject = b;
+  });
+  return {
+    promise,
+    resolve,
+    reject
+  };
+}

packages/node/tests/utils.ts

@@ -14,6 +14,7 @@ import {
   PowerSyncBackendConnector,
   PowerSyncCredentials,
   PowerSyncDatabase,
+  PowerSyncDatabaseOptions,
   Schema,
   StreamingSyncCheckpoint,
   StreamingSyncLine,
@@ -65,27 +66,32 @@ export async function createDatabase(
 
   const database = new PowerSyncDatabase({
     schema: AppSchema,
+    ...options,
+    logger: defaultLogger,
     database: {
       dbFilename: 'test.db',
       dbLocation: tmpdir,
       // Using a single read worker (instead of multiple, the default) seems to improve the reliability of tests in GH
       // actions. So far, we've not been able to reproduce these failures locally.
-      readWorkerCount: 1
-    },
-    logger: defaultLogger,
-    ...options
+      readWorkerCount: 1,
+      ...options.database
+    }
   });
   await database.init();
   return database;
 }
 
-export const databaseTest = tempDirectoryTest.extend<{ database: PowerSyncDatabase }>({
-  database: async ({ tmpdir }, use) => {
-    const db = await createDatabase(tmpdir);
-    await use(db);
-    await db.close();
-  }
-});
+export const customDatabaseTest = (options?: Partial<NodePowerSyncDatabaseOptions>) => {
+  return tempDirectoryTest.extend<{ database: PowerSyncDatabase }>({
+    database: async ({ tmpdir }, use) => {
+      const db = await createDatabase(tmpdir, options);
+      await use(db);
+      await db.close();
+    }
+  });
+};
+
+export const databaseTest = customDatabaseTest();
 
 // TODO: Unify this with the test setup for the web SDK.
 export const mockSyncServiceTest = tempDirectoryTest.extend<{