diff --git a/packages/powersync_core/lib/src/database/native/native_powersync_database.dart b/packages/powersync_core/lib/src/database/native/native_powersync_database.dart
index 2f846fd5..94e3669b 100644
--- a/packages/powersync_core/lib/src/database/native/native_powersync_database.dart
+++ b/packages/powersync_core/lib/src/database/native/native_powersync_database.dart
@@ -1,4 +1,5 @@
 import 'dart:async';
+import 'dart:convert';
 import 'dart:isolate';
 import 'package:meta/meta.dart';
 
@@ -44,6 +45,9 @@ class PowerSyncDatabaseImpl
   @override
   SqliteDatabase database;
 
+  @override
+  bool manualSchemaManagement;
+
   @override
   @protected
   late Future<void> isInitialized;
@@ -76,6 +80,7 @@ class PowerSyncDatabaseImpl
       required String path,
       int maxReaders = SqliteDatabase.defaultMaxReaders,
       Logger? logger,
+      bool manualSchemaManagement = false,
       @Deprecated("Use [PowerSyncDatabase.withFactory] instead.")
       // ignore: deprecated_member_use_from_same_package
       SqliteConnectionSetup? sqliteSetup}) {
@@ -83,8 +88,13 @@ class PowerSyncDatabaseImpl
     DefaultSqliteOpenFactory factory =
         // ignore: deprecated_member_use_from_same_package
         PowerSyncOpenFactory(path: path, sqliteSetup: sqliteSetup);
-    return PowerSyncDatabaseImpl.withFactory(factory,
-        schema: schema, maxReaders: maxReaders, logger: logger);
+    return PowerSyncDatabaseImpl.withFactory(
+      factory,
+      schema: schema,
+      maxReaders: maxReaders,
+      logger: logger,
+      manualSchemaManagement: manualSchemaManagement,
+    );
   }
 
   /// Open a [PowerSyncDatabase] with a [PowerSyncOpenFactory].
@@ -96,13 +106,19 @@ class PowerSyncDatabaseImpl
   ///
   /// [logger] defaults to [autoLogger], which logs to the console in debug builds.
   factory PowerSyncDatabaseImpl.withFactory(
-      DefaultSqliteOpenFactory openFactory,
-      {required Schema schema,
-      int maxReaders = SqliteDatabase.defaultMaxReaders,
-      Logger? logger}) {
+    DefaultSqliteOpenFactory openFactory, {
+    required Schema schema,
+    int maxReaders = SqliteDatabase.defaultMaxReaders,
+    Logger? logger,
+    bool manualSchemaManagement = false,
+  }) {
     final db = SqliteDatabase.withFactory(openFactory, maxReaders: maxReaders);
     return PowerSyncDatabaseImpl.withDatabase(
-        schema: schema, database: db, logger: logger);
+      schema: schema,
+      database: db,
+      logger: logger,
+      manualSchemaManagement: manualSchemaManagement,
+    );
   }
 
   /// Open a PowerSyncDatabase on an existing [SqliteDatabase].
@@ -110,8 +126,12 @@ class PowerSyncDatabaseImpl
   /// Migrations are run on the database when this constructor is called.
   ///
   /// [logger] defaults to [autoLogger], which logs to the console in debug builds.s
-  PowerSyncDatabaseImpl.withDatabase(
-      {required this.schema, required this.database, Logger? logger}) {
+  PowerSyncDatabaseImpl.withDatabase({
+    required this.schema,
+    required this.database,
+    Logger? logger,
+    this.manualSchemaManagement = false,
+  }) {
     this.logger = logger ?? autoLogger;
     isInitialized = baseInit();
   }
@@ -247,6 +267,7 @@ class PowerSyncDatabaseImpl
         options,
         crudMutex.shared,
         syncMutex.shared,
+        jsonEncode(schema),
       ),
       debugName: 'Sync ${database.openFactory.path}',
       onError: receiveUnhandledErrors.sendPort,
@@ -290,6 +311,7 @@ class _PowerSyncDatabaseIsolateArgs {
   final ResolvedSyncOptions options;
   final SerializedMutex crudMutex;
   final SerializedMutex syncMutex;
+  final String schemaJson;
 
   _PowerSyncDatabaseIsolateArgs(
     this.sPort,
@@ -297,6 +319,7 @@ class _PowerSyncDatabaseIsolateArgs {
     this.options,
     this.crudMutex,
     this.syncMutex,
+    this.schemaJson,
   );
 }
 
@@ -392,6 +415,7 @@ Future<void> _syncIsolate(_PowerSyncDatabaseIsolateArgs args) async {
     final storage = BucketStorage(connection);
     final sync = StreamingSyncImplementation(
       adapter: storage,
+      schemaJson: args.schemaJson,
       connector: InternalConnector(
         getCredentialsCached: getCredentialsCached,
         prefetchCredentials: prefetchCredentials,
diff --git a/packages/powersync_core/lib/src/database/powersync_database.dart b/packages/powersync_core/lib/src/database/powersync_database.dart
index 95543ce8..eb220758 100644
--- a/packages/powersync_core/lib/src/database/powersync_database.dart
+++ b/packages/powersync_core/lib/src/database/powersync_database.dart
@@ -32,19 +32,23 @@ abstract class PowerSyncDatabase
   /// A maximum of [maxReaders] concurrent read transactions are allowed.
   ///
   /// [logger] defaults to [autoLogger], which logs to the console in debug builds.
-  factory PowerSyncDatabase(
-      {required Schema schema,
-      required String path,
-      Logger? logger,
-      @Deprecated("Use [PowerSyncDatabase.withFactory] instead.")
-      // ignore: deprecated_member_use_from_same_package
-      SqliteConnectionSetup? sqliteSetup}) {
+  factory PowerSyncDatabase({
+    required Schema schema,
+    required String path,
+    Logger? logger,
+    bool manualSchemaManagement = false,
+    @Deprecated("Use [PowerSyncDatabase.withFactory] instead.")
+    // ignore: deprecated_member_use_from_same_package
+    SqliteConnectionSetup? sqliteSetup,
+  }) {
     return PowerSyncDatabaseImpl(
-        schema: schema,
-        path: path,
-        logger: logger,
-        // ignore: deprecated_member_use_from_same_package
-        sqliteSetup: sqliteSetup);
+      schema: schema,
+      path: path,
+      manualSchemaManagement: manualSchemaManagement,
+      logger: logger,
+      // ignore: deprecated_member_use_from_same_package
+      sqliteSetup: sqliteSetup,
+    );
   }
 
   /// Open a [PowerSyncDatabase] with a [PowerSyncOpenFactory].
@@ -55,12 +59,20 @@ abstract class PowerSyncDatabase
   /// Subclass [PowerSyncOpenFactory] to add custom logic to this process.
   ///
   /// [logger] defaults to [autoLogger], which logs to the console in debug builds.
-  factory PowerSyncDatabase.withFactory(DefaultSqliteOpenFactory openFactory,
-      {required Schema schema,
-      int maxReaders = SqliteDatabase.defaultMaxReaders,
-      Logger? logger}) {
-    return PowerSyncDatabaseImpl.withFactory(openFactory,
-        schema: schema, maxReaders: maxReaders, logger: logger);
+  factory PowerSyncDatabase.withFactory(
+    DefaultSqliteOpenFactory openFactory, {
+    required Schema schema,
+    int maxReaders = SqliteDatabase.defaultMaxReaders,
+    bool manualSchemaManagement = false,
+    Logger? logger,
+  }) {
+    return PowerSyncDatabaseImpl.withFactory(
+      openFactory,
+      schema: schema,
+      maxReaders: maxReaders,
+      manualSchemaManagement: manualSchemaManagement,
+      logger: logger,
+    );
   }
 
   /// Open a PowerSyncDatabase on an existing [SqliteDatabase].
@@ -68,11 +80,17 @@ abstract class PowerSyncDatabase
   /// Migrations are run on the database when this constructor is called.
   ///
   /// [logger] defaults to [autoLogger], which logs to the console in debug builds.
-  factory PowerSyncDatabase.withDatabase(
-      {required Schema schema,
-      required SqliteDatabase database,
-      Logger? loggers}) {
+  factory PowerSyncDatabase.withDatabase({
+    required Schema schema,
+    required SqliteDatabase database,
+    bool manualSchemaManagement = false,
+    Logger? logger,
+  }) {
     return PowerSyncDatabaseImpl.withDatabase(
-        schema: schema, database: database, logger: loggers);
+      schema: schema,
+      database: database,
+      manualSchemaManagement: manualSchemaManagement,
+      logger: logger,
+    );
   }
 }
diff --git a/packages/powersync_core/lib/src/database/powersync_database_impl_stub.dart b/packages/powersync_core/lib/src/database/powersync_database_impl_stub.dart
index 2a795497..ee3ab2af 100644
--- a/packages/powersync_core/lib/src/database/powersync_database_impl_stub.dart
+++ b/packages/powersync_core/lib/src/database/powersync_database_impl_stub.dart
@@ -32,6 +32,9 @@ class PowerSyncDatabaseImpl
   @override
   SqliteDatabase get database => throw UnimplementedError();
 
+  @override
+  bool get manualSchemaManagement => throw UnimplementedError();
+
   @override
   Future<void> get isInitialized => throw UnimplementedError();
 
@@ -53,6 +56,7 @@ class PowerSyncDatabaseImpl
       {required Schema schema,
       required String path,
       int maxReaders = SqliteDatabase.defaultMaxReaders,
+      bool manualSchemaManagement = false,
       Logger? logger,
       @Deprecated("Use [PowerSyncDatabase.withFactory] instead.")
       // ignore: deprecated_member_use_from_same_package
@@ -72,6 +76,7 @@ class PowerSyncDatabaseImpl
     DefaultSqliteOpenFactory openFactory, {
     required Schema schema,
     int maxReaders = SqliteDatabase.defaultMaxReaders,
+    bool manualSchemaManagement = false,
     Logger? logger,
   }) {
     throw UnimplementedError();
@@ -82,10 +87,12 @@ class PowerSyncDatabaseImpl
   /// Migrations are run on the database when this constructor is called.
   ///
   /// [logger] defaults to [autoLogger], which logs to the console in debug builds.s
-  factory PowerSyncDatabaseImpl.withDatabase(
-      {required Schema schema,
-      required SqliteDatabase database,
-      Logger? logger}) {
+  factory PowerSyncDatabaseImpl.withDatabase({
+    required Schema schema,
+    required SqliteDatabase database,
+    bool manualSchemaManagement = false,
+    Logger? logger,
+  }) {
     throw UnimplementedError();
   }
 
diff --git a/packages/powersync_core/lib/src/database/powersync_db_mixin.dart b/packages/powersync_core/lib/src/database/powersync_db_mixin.dart
index 808efc71..64744797 100644
--- a/packages/powersync_core/lib/src/database/powersync_db_mixin.dart
+++ b/packages/powersync_core/lib/src/database/powersync_db_mixin.dart
@@ -38,6 +38,10 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
   /// Use [attachedLogger] to propagate logs to [Logger.root] for custom logging.
   Logger get logger;
 
+  bool get manualSchemaManagement;
+
+  bool _manualSchemaManagementCompleted = false;
+
   @Deprecated("This field is unused, pass params to connect() instead")
   Map<String, dynamic>? clientParams;
 
@@ -110,10 +114,36 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
     statusStream = statusStreamController.stream;
     updates = powerSyncUpdateNotifications(database.updates);
 
+    _manualSchemaManagementCompleted = false;
+
     await database.initialize();
     await _checkVersion();
     await database.execute('SELECT powersync_init()');
-    await updateSchema(schema);
+
+    if (!manualSchemaManagement) {
+      // Create the internal db schema
+      await updateSchema(schema);
+      await _afterSchemaReady();
+    }
+  }
+
+  Future<void> markSchemaAsReady() async {
+    await isInitialized;
+    _manualSchemaManagementCompleted = true;
+
+    await _afterSchemaReady();
+  }
+
+  void _checkSchemaIsReady() {
+    if (!manualSchemaManagement || _manualSchemaManagementCompleted) {
+      return;
+    }
+
+    throw StateError(
+        'In manual schema management mode, you need to mark the powersync database as ready');
+  }
+
+  Future<void> _afterSchemaReady() async {
     await _updateHasSynced();
   }
 
@@ -289,6 +319,8 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
     // the lock for the connection.
     await initialize();
 
+    _checkSchemaIsReady();
+
     final resolvedOptions = ResolvedSyncOptions.resolve(
       options,
       crudThrottleTime: crudThrottleTime,
@@ -452,6 +484,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
   /// Get an unique id for this client.
   /// This id is only reset when the database is deleted.
   Future<String> getClientId() async {
+    _checkSchemaIsReady(); // TODO(skilldevs): Needed?
     final row = await get('SELECT powersync_client_id() as client_id');
     return row['client_id'] as String;
   }
@@ -459,6 +492,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
   /// Get upload queue size estimate and count.
   Future<UploadQueueStats> getUploadQueueStats(
       {bool includeSize = false}) async {
+    _checkSchemaIsReady();
     if (includeSize) {
       final row = await getOptional(
           'SELECT SUM(cast(data as blob) + 20) as size, count(*) as count FROM ps_crud');
@@ -486,6 +520,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
   /// data by transaction. One batch may contain data from multiple transactions,
   /// and a single transaction may be split over multiple batches.
   Future<CrudBatch?> getCrudBatch({int limit = 100}) async {
+    _checkSchemaIsReady();
     final rows = await getAll(
         'SELECT id, tx_id, data FROM ps_crud ORDER BY id ASC LIMIT ?',
         [limit + 1]);
@@ -532,6 +567,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
   /// Unlike [getCrudBatch], this only returns data from a single transaction at a time.
   /// All data for the transaction is loaded into memory.
   Future<CrudTransaction?> getNextCrudTransaction() async {
+    _checkSchemaIsReady();
     return await readTransaction((tx) async {
       final first = await tx.getOptional(
           'SELECT id, tx_id, data FROM ps_crud ORDER BY id ASC LIMIT 1');
diff --git a/packages/powersync_core/lib/src/database/web/web_powersync_database.dart b/packages/powersync_core/lib/src/database/web/web_powersync_database.dart
index 6b40a6a2..a879d9e0 100644
--- a/packages/powersync_core/lib/src/database/web/web_powersync_database.dart
+++ b/packages/powersync_core/lib/src/database/web/web_powersync_database.dart
@@ -1,4 +1,5 @@
 import 'dart:async';
+import 'dart:convert';
 import 'package:meta/meta.dart';
 import 'package:http/browser_client.dart';
 import 'package:logging/logging.dart';
@@ -38,6 +39,9 @@ class PowerSyncDatabaseImpl
   @override
   SqliteDatabase database;
 
+  @override
+  bool manualSchemaManagement;
+
   @override
   @protected
   late Future<void> isInitialized;
@@ -69,14 +73,20 @@ class PowerSyncDatabaseImpl
       {required Schema schema,
       required String path,
       int maxReaders = SqliteDatabase.defaultMaxReaders,
+      bool manualSchemaManagement = false,
       Logger? logger,
       @Deprecated("Use [PowerSyncDatabase.withFactory] instead.")
       // ignore: deprecated_member_use_from_same_package
       SqliteConnectionSetup? sqliteSetup}) {
     // ignore: deprecated_member_use_from_same_package
     DefaultSqliteOpenFactory factory = PowerSyncOpenFactory(path: path);
-    return PowerSyncDatabaseImpl.withFactory(factory,
-        maxReaders: maxReaders, logger: logger, schema: schema);
+    return PowerSyncDatabaseImpl.withFactory(
+      factory,
+      maxReaders: maxReaders,
+      logger: logger,
+      schema: schema,
+      manualSchemaManagement: manualSchemaManagement,
+    );
   }
 
   /// Open a [PowerSyncDatabase] with a [PowerSyncOpenFactory].
@@ -91,10 +101,15 @@ class PowerSyncDatabaseImpl
       DefaultSqliteOpenFactory openFactory,
       {required Schema schema,
       int maxReaders = SqliteDatabase.defaultMaxReaders,
+      bool manualSchemaManagement = false,
       Logger? logger}) {
     final db = SqliteDatabase.withFactory(openFactory, maxReaders: 1);
     return PowerSyncDatabaseImpl.withDatabase(
-        schema: schema, logger: logger, database: db);
+      schema: schema,
+      manualSchemaManagement: manualSchemaManagement,
+      logger: logger,
+      database: db,
+    );
   }
 
   /// Open a PowerSyncDatabase on an existing [SqliteDatabase].
@@ -102,8 +117,12 @@ class PowerSyncDatabaseImpl
   /// Migrations are run on the database when this constructor is called.
   ///
   /// [logger] defaults to [autoLogger], which logs to the console in debug builds.
-  PowerSyncDatabaseImpl.withDatabase(
-      {required this.schema, required this.database, Logger? logger}) {
+  PowerSyncDatabaseImpl.withDatabase({
+    required this.schema,
+    required this.database,
+    this.manualSchemaManagement = false,
+    Logger? logger,
+  }) {
     if (logger != null) {
       this.logger = logger;
     } else {
@@ -141,6 +160,7 @@ class PowerSyncDatabaseImpl
 
       sync = StreamingSyncImplementation(
         adapter: storage,
+        schemaJson: jsonEncode(schema),
         connector: InternalConnector.wrap(connector, this),
         crudUpdateTriggerStream: crudStream,
         options: options,
diff --git a/packages/powersync_core/lib/src/schema.dart b/packages/powersync_core/lib/src/schema.dart
index 4892ee6c..d5273c38 100644
--- a/packages/powersync_core/lib/src/schema.dart
+++ b/packages/powersync_core/lib/src/schema.dart
@@ -8,10 +8,11 @@ import 'schema_logic.dart';
 class Schema {
   /// List of tables in the schema.
   final List<Table> tables;
+  final List<RawTable> rawTables;
 
-  const Schema(this.tables);
+  const Schema(this.tables, {this.rawTables = const []});
 
-  Map<String, dynamic> toJson() => {'tables': tables};
+  Map<String, dynamic> toJson() => {'raw_tables': rawTables, 'tables': tables};
 
   void validate() {
     Set<String> tableNames = {};
@@ -315,6 +316,64 @@ class Column {
   Map<String, dynamic> toJson() => {'name': name, 'type': type.sqlite};
 }
 
+final class RawTable {
+  final String name;
+  final PendingStatement put;
+  final PendingStatement delete;
+
+  const RawTable({
+    required this.name,
+    required this.put,
+    required this.delete,
+  });
+
+  Map<String, dynamic> toJson() => {
+        'name': name,
+        'put': put,
+        'delete': delete,
+      };
+}
+
+final class PendingStatement {
+  final String sql;
+  final List<PendingStatementValue> params;
+
+  PendingStatement({required this.sql, required this.params});
+
+  Map<String, dynamic> toJson() => {
+        'sql': sql,
+        'params': params,
+      };
+}
+
+sealed class PendingStatementValue {
+  factory PendingStatementValue.id() = _PendingStmtValueId;
+  factory PendingStatementValue.column(String column) = _PendingStmtValueColumn;
+
+  dynamic toJson();
+}
+
+class _PendingStmtValueColumn implements PendingStatementValue {
+  final String column;
+  const _PendingStmtValueColumn(this.column);
+
+  @override
+  dynamic toJson() {
+    return {
+      'Column': column,
+    };
+  }
+}
+
+class _PendingStmtValueId implements PendingStatementValue {
+  const _PendingStmtValueId();
+
+  @override
+  dynamic toJson() {
+    return 'Id';
+  }
+}
+
 /// Type of column.
 enum ColumnType {
   /// TEXT column.
diff --git a/packages/powersync_core/lib/src/sync/streaming_sync.dart b/packages/powersync_core/lib/src/sync/streaming_sync.dart
index b2e2f5bc..b4dbe068 100644
--- a/packages/powersync_core/lib/src/sync/streaming_sync.dart
+++ b/packages/powersync_core/lib/src/sync/streaming_sync.dart
@@ -32,6 +32,7 @@ abstract interface class StreamingSync {
 
 @internal
 class StreamingSyncImplementation implements StreamingSync {
+  final String schemaJson;
   final BucketStorage adapter;
   final InternalConnector connector;
   final ResolvedSyncOptions options;
@@ -62,6 +63,7 @@ class StreamingSyncImplementation implements StreamingSync {
   String? clientId;
 
   StreamingSyncImplementation({
+    required this.schemaJson,
     required this.adapter,
     required this.connector,
     required this.crudUpdateTriggerStream,
@@ -592,7 +594,7 @@ final class _ActiveRustStreamingIteration {
         'start',
         convert.json.encode({
           'parameters': sync.options.params,
-          'schema': 'TODO: Pass-through schema (probably in serialized form)',
+          'schema': convert.json.decode(sync.schemaJson),
         }),
       );
       assert(_completedStream.isCompleted, 'Should have started streaming');
diff --git a/packages/powersync_core/lib/src/web/sync_controller.dart b/packages/powersync_core/lib/src/web/sync_controller.dart
index 0c26252e..7f05cff3 100644
--- a/packages/powersync_core/lib/src/web/sync_controller.dart
+++ b/packages/powersync_core/lib/src/web/sync_controller.dart
@@ -113,6 +113,9 @@ class SyncWorkerHandle implements StreamingSync {
   @override
   Future<void> streamingSync() async {
     await _channel.startSynchronization(
-        database.database.openFactory.path, ResolvedSyncOptions(options));
+      database.database.openFactory.path,
+      ResolvedSyncOptions(options),
+      database.schema,
+    );
   }
 }
diff --git a/packages/powersync_core/lib/src/web/sync_worker.dart b/packages/powersync_core/lib/src/web/sync_worker.dart
index b5e8ed63..ddc4eaf0 100644
--- a/packages/powersync_core/lib/src/web/sync_worker.dart
+++ b/packages/powersync_core/lib/src/web/sync_worker.dart
@@ -45,12 +45,16 @@ class _SyncWorker {
     });
   }
 
-  _SyncRunner referenceSyncTask(
-      String databaseIdentifier, SyncOptions options, _ConnectedClient client) {
+  _SyncRunner referenceSyncTask(String databaseIdentifier, SyncOptions options,
+      String schemaJson, _ConnectedClient client) {
     return _requestedSyncTasks.putIfAbsent(databaseIdentifier, () {
       return _SyncRunner(databaseIdentifier);
     })
-      ..registerClient(client, options);
+      ..registerClient(
+        client,
+        options,
+        schemaJson,
+      );
   }
 }
 
@@ -86,8 +90,8 @@ class _ConnectedClient {
               },
             );
 
-            _runner = _worker.referenceSyncTask(
-                request.databaseName, recoveredOptions, this);
+            _runner = _worker.referenceSyncTask(request.databaseName,
+                recoveredOptions, request.schemaJson, this);
             return (JSObject(), null);
           case SyncWorkerMessageType.abortSynchronization:
             _runner?.disconnectClient(this);
@@ -128,6 +132,7 @@ class _ConnectedClient {
 class _SyncRunner {
   final String identifier;
   ResolvedSyncOptions options = ResolvedSyncOptions(SyncOptions());
+  String schemaJson = '{}';
 
   final StreamGroup<_RunnerEvent> _group = StreamGroup();
   final StreamController<_RunnerEvent> _mainEvents = StreamController();
@@ -146,10 +151,12 @@ class _SyncRunner {
             case _AddConnection(
                 :final client,
                 :final options,
+                :final schemaJson,
               ):
               connections.add(client);
               final (newOptions, reconnect) = this.options.applyFrom(options);
               this.options = newOptions;
+              this.schemaJson = schemaJson;
 
               if (sync == null) {
                 await _requestDatabase(client);
@@ -264,6 +271,7 @@ class _SyncRunner {
 
     sync = StreamingSyncImplementation(
       adapter: WebBucketStorage(database),
+      schemaJson: client._runner!.schemaJson,
       connector: InternalConnector(
         getCredentialsCached: client.channel.credentialsCallback,
         prefetchCredentials: ({required bool invalidate}) async {
@@ -286,8 +294,9 @@ class _SyncRunner {
     sync!.streamingSync();
   }
 
-  void registerClient(_ConnectedClient client, SyncOptions options) {
-    _mainEvents.add(_AddConnection(client, options));
+  void registerClient(
+      _ConnectedClient client, SyncOptions options, String schemaJson) {
+    _mainEvents.add(_AddConnection(client, options, schemaJson));
   }
 
   /// Remove a client, disconnecting if no clients remain..
@@ -306,8 +315,9 @@ sealed class _RunnerEvent {}
 final class _AddConnection implements _RunnerEvent {
   final _ConnectedClient client;
   final SyncOptions options;
+  final String schemaJson;
 
-  _AddConnection(this.client, this.options);
+  _AddConnection(this.client, this.options, this.schemaJson);
 }
 
 final class _RemoveConnection implements _RunnerEvent {
diff --git a/packages/powersync_core/lib/src/web/sync_worker_protocol.dart b/packages/powersync_core/lib/src/web/sync_worker_protocol.dart
index 2b859e53..3c64d90f 100644
--- a/packages/powersync_core/lib/src/web/sync_worker_protocol.dart
+++ b/packages/powersync_core/lib/src/web/sync_worker_protocol.dart
@@ -3,6 +3,7 @@ import 'dart:convert';
 import 'dart:js_interop';
 
 import 'package:logging/logging.dart';
+import 'package:powersync_core/src/schema.dart';
 import 'package:powersync_core/src/sync/options.dart';
 import 'package:web/web.dart';
 
@@ -71,6 +72,7 @@ extension type StartSynchronization._(JSObject _) implements JSObject {
     required int requestId,
     required int retryDelayMs,
     required String implementationName,
+    required String schemaJson,
     String? syncParamsEncoded,
   });
 
@@ -79,6 +81,7 @@ extension type StartSynchronization._(JSObject _) implements JSObject {
   external int get crudThrottleTimeMs;
   external int? get retryDelayMs;
   external String? get implementationName;
+  external String get schemaJson;
   external String? get syncParamsEncoded;
 }
 
@@ -410,7 +413,7 @@ final class WorkerCommunicationChannel {
   }
 
   Future<void> startSynchronization(
-      String databaseName, ResolvedSyncOptions options) async {
+      String databaseName, ResolvedSyncOptions options, Schema schema) async {
     final (id, completion) = _newRequest();
     port.postMessage(SyncWorkerMessage(
       type: SyncWorkerMessageType.startSynchronization.name,
@@ -420,6 +423,7 @@ final class WorkerCommunicationChannel {
         retryDelayMs: options.retryDelay.inMilliseconds,
         requestId: id,
         implementationName: options.source.syncImplementation.name,
+        schemaJson: jsonEncode(schema),
         syncParamsEncoded: switch (options.source.params) {
           null => null,
           final params => jsonEncode(params),
diff --git a/packages/powersync_core/test/in_memory_sync_test.dart b/packages/powersync_core/test/in_memory_sync_test.dart
index 9455bc94..c735d817 100644
--- a/packages/powersync_core/test/in_memory_sync_test.dart
+++ b/packages/powersync_core/test/in_memory_sync_test.dart
@@ -1,4 +1,5 @@
 import 'dart:async';
+import 'dart:convert';
 
 import 'package:async/async.dart';
 import 'package:logging/logging.dart';
@@ -172,7 +173,7 @@ void _declareTests(String name, SyncOptions options) {
                   'object_type': 'a',
                   'object_id': '1',
                   'checksum': 0,
-                  'data': {},
+                  'data': '{}',
                 }
               ],
             }
@@ -187,7 +188,7 @@ void _declareTests(String name, SyncOptions options) {
                   'object_type': 'b',
                   'object_id': '1',
                   'checksum': 0,
-                  'data': {},
+                  'data': '{}',
                 }
               ],
             }
@@ -229,7 +230,7 @@ void _declareTests(String name, SyncOptions options) {
               'data': [
                 {
                   'checksum': priority + 10,
-                  'data': {'name': 'test', 'email': 'email'},
+                  'data': json.encode({'name': 'test', 'email': 'email'}),
                   'op': 'PUT',
                   'op_id': '${operationId++}',
                   'object_id': 'prio$priority',
@@ -411,7 +412,7 @@ void _declareTests(String name, SyncOptions options) {
             'data': [
               {
                 'checksum': 0,
-                'data': {'name': 'from local', 'email': 'local@example.org'},
+                'data': json.encode({'name': 'from local', 'email': 'local@example.org'}),
                 'op': 'PUT',
                 'op_id': '1',
                 'object_id': '1',
@@ -419,7 +420,7 @@ void _declareTests(String name, SyncOptions options) {
               },
               {
                 'checksum': 0,
-                'data': {'name': 'additional', 'email': ''},
+                'data': json.encode({'name': 'additional', 'email': ''}),
                 'op': 'PUT',
                 'op_id': '2',
                 'object_id': '2',
@@ -477,7 +478,7 @@ void _declareTests(String name, SyncOptions options) {
                   'object_type': bucket,
                   'object_id': '$lastOpId',
                   'checksum': 0,
-                  'data': {},
+                  'data': '{}',
                 }
             ],
           }
diff --git a/packages/powersync_core/test/utils/abstract_test_utils.dart b/packages/powersync_core/test/utils/abstract_test_utils.dart
index 3ea4a319..f402bb5f 100644
--- a/packages/powersync_core/test/utils/abstract_test_utils.dart
+++ b/packages/powersync_core/test/utils/abstract_test_utils.dart
@@ -1,3 +1,5 @@
+import 'dart:convert';
+
 import 'package:http/http.dart';
 import 'package:logging/logging.dart';
 import 'package:powersync_core/powersync_core.dart';
@@ -74,7 +76,7 @@ abstract mixin class TestPowerSyncFactory implements PowerSyncOpenFactory {
       schema: schema,
       database: SqliteDatabase.singleConnection(
           SqliteConnection.synchronousWrapper(raw)),
-      loggers: logger,
+      logger: logger,
     );
   }
 }
@@ -153,6 +155,7 @@ extension MockSync on PowerSyncDatabase {
   }) {
     final impl = StreamingSyncImplementation(
       adapter: BucketStorage(this),
+      schemaJson: jsonEncode(schema),
       client: client,
       options: ResolvedSyncOptions(options),
       connector: InternalConnector.wrap(connector, this),