Support direct tables

packages/powersync/CHANGELOG.md

@@ -4,6 +4,7 @@
 - The DevTools extension is now based on the shared [PowerSync DevTools](https://releases.powersync.com/announcements/powersync-devtools-inspect-your-app-s-live-database). This improves the SQL inspector, reports per-bucket progress, allows subscribing to Sync Streams and has a cleaner UI.
 - Fix `disconnect()`, `disconnectAndClear()` and `close()` hanging while a `requestCheckpoint()` call waits for an
   unresponsive service. Pending checkpoint requests now fail with a `CheckpointRequestException` on disconnect.
+- Support [direct tables](https://docs.powersync.com/client-sdks/advanced/performance-optimization#direct-tables).
 
 ## 2.5.0
 

packages/powersync/lib/src/schema.dart

@@ -1,6 +1,8 @@
 /// @docImport 'database/powersync_database.dart';
 library;
 
+import 'package:meta/meta.dart';
+
 import 'crud.dart';
 import 'schema_logic.dart';
 
@@ -138,6 +140,8 @@ base class Table extends TableOptions {
   /// Override the name for the view
   final String? _viewNameOverride;
 
+  final bool _isDirect;
+
   /// powersync-sqlite-core limits the number of columns
   /// per table to 1999, due to internal SQLite limits.
   ///
@@ -168,6 +172,7 @@ base class Table extends TableOptions {
     super.trackMetadata,
     super.trackPreviousValues,
   }) : _viewNameOverride = viewName,
+       _isDirect = false,
        super(insertOnly: false);
 
   /// Create a table that only exists locally.
@@ -179,6 +184,7 @@ base class Table extends TableOptions {
     this.indexes = const [],
     String? viewName,
   }) : _viewNameOverride = viewName,
+       _isDirect = false,
        super(localOnly: true);
 
   /// Create a table that only supports inserts.
@@ -198,8 +204,32 @@ base class Table extends TableOptions {
     super.trackPreviousValues,
   }) : indexes = const [],
        _viewNameOverride = viewName,
+       _isDirect = false,
        super(localOnly: false, insertOnly: true);
 
+  /// Creates a direct table, storing data directly instead of through an
+  /// intermediate view around a JSON table.
+  ///
+  /// Direct tables are also managed by the PowerSync SDK, which is responsible
+  /// for creating, altering and dropping them.
+  ///
+  /// You can query raw tables like JSON-based tables, and mutations are
+  /// consistently tracked as pending uploads.
+  ///
+  /// For more information on direct tables, see [the documentation](https://docs.powersync.com/client-sdks/advanced/performance-optimization#direct-tables).
+  @experimental
+  const Table.direct(
+    this.name,
+    this.columns, {
+    this.indexes = const [],
+    super.trackMetadata,
+    super.trackPreviousValues,
+    super.localOnly,
+    super.insertOnly,
+    super.ignoreEmptyUpdates,
+  }) : _isDirect = true,
+       _viewNameOverride = null;
+
   Column operator [](String columnName) {
     return columns.firstWhere((element) => element.name == columnName);
   }
@@ -281,6 +311,7 @@ base class Table extends TableOptions {
     'view_name': _viewNameOverride,
     'columns': columns,
     'indexes': indexes.map((e) => e.toJson(this)).toList(growable: false),
+    if (_isDirect) 'direct': true,
     ..._optionsToJson(),
   };
 }

packages/powersync/test/sync/in_memory_sync_test.dart

@@ -348,6 +348,60 @@ void _declareTests(String name, bool bson) {
       await expectLater(query, emits(isEmpty));
     });
 
+    test('direct tables', () async {
+      await database.close();
+      (_, database) = await testUtils.openInMemoryDatabase(
+        schema: Schema([
+          .direct('users', [.text('name')]),
+        ]),
+      );
+
+      await database.initialize();
+      expect(
+        await database.get('pragma table_list(users)'),
+        containsPair('type', 'table'),
+      );
+
+      await waitForConnection();
+      syncService
+        ..addLine(
+          checkpoint(lastOpId: 1, buckets: [bucketDescription('bucket')]),
+        )
+        ..addLine({
+          'data': {
+            'bucket': 'bucket',
+            'data': [
+              {
+                'checksum': 0,
+                'data': json.encode({'name': 'synced user'}),
+                'op': 'PUT',
+                'op_id': '1',
+                'object_id': 'user_id',
+                'object_type': 'users',
+              },
+            ],
+          },
+        })
+        ..addLine(checkpointComplete());
+      await database.waitForFirstSync();
+
+      expect(await database.get('SELECT * FROM users'), {
+        'id': 'user_id',
+        'name': 'synced user',
+      });
+      await database.disconnect();
+
+      await database.execute('UPDATE users SET name = ?', ['local user']);
+      expect(await database.get('SELECT * FROM users'), {
+        'id': 'user_id',
+        'name': 'local user',
+      });
+      final CrudTransaction(crud: [entry]) = (await database
+          .getNextCrudTransaction())!;
+      expect(entry.table, 'users');
+      expect(entry.opData, {'name': 'local user'});
+    });
+
     test('marks as connected even without sync line', () async {
       await waitForConnection(addKeepLive: false);
       expect(database.currentStatus.connected, isTrue);

packages/powersync/test/sync/utils.dart

@@ -11,6 +11,7 @@ TypeMatcher<SyncStatus> isSyncStatus({
   Object? downloadProgress,
   Object? syncStreams,
   Object? downloadError,
+  Object? uploading,
 }) {
   var matcher = isA<SyncStatus>();
   if (downloading != null) {
@@ -43,6 +44,10 @@ TypeMatcher<SyncStatus> isSyncStatus({
     );
   }
 
+  if (uploading != null) {
+    matcher = matcher.having((e) => e.uploading, 'uploading', uploading);
+  }
+
   return matcher;
 }
 

packages/powersync/test/web/sync_worker_test.dart

@@ -4,6 +4,7 @@ library;
 import 'dart:async';
 import 'dart:convert';
 
+import 'package:async/async.dart';
 import 'package:logging/logging.dart';
 import 'package:powersync/powersync.dart';
 import 'package:powersync/src/sync/streaming_sync.dart';
@@ -129,15 +130,23 @@ void main() {
       connector: _ThrowingBackendConnector(),
       options: options,
     );
+    final statusChanges = StreamQueue(db.statusStream);
     await first.streamingSync();
-    await db.statusStream.firstWhere((s) => s.downloadError != null);
+
+    await expectLater(
+      statusChanges,
+      emitsThrough(isSyncStatus(downloadError: isNotNull)),
+    );
 
     // Trigger an upload, emitting status updates that don't change the
     // download error.
     await db.execute(
       "INSERT INTO customers (id, name) VALUES (uuid(), 'name')",
     );
-    await db.statusStream.firstWhere((s) => s.uploading);
+    await expectLater(
+      statusChanges,
+      emitsThrough(isSyncStatus(uploading: true)),
+    );
 
     // A connecting client should receive the existing download error.
     final second = createWorkerHandle(
@@ -156,6 +165,7 @@ void main() {
       contains('Expected error from fetchCredentials'),
     );
 
+    await statusChanges.cancel();
     await first.abort();
   });