[MySQL] Automatic schema change handling

.changeset/wet-berries-enjoy.md

@@ -0,0 +1,21 @@
+---
+'@powersync/service-module-postgres-storage': minor
+'@powersync/service-module-mongodb-storage': minor
+'@powersync/service-core-tests': minor
+'@powersync/service-module-postgres': minor
+'@powersync/service-module-mongodb': minor
+'@powersync/service-core': minor
+'@powersync/service-module-mysql': minor
+'@powersync/service-sync-rules': minor
+---
+
+MySQL: 
+- Added schema change handling
+ - Except for some edge cases, the following schema changes are now handled automatically:
+   - Creation, renaming, dropping and truncation of tables.
+   - Creation and dropping of unique indexes and primary keys.
+   - Adding, modifying, dropping and renaming of table columns.
+ - If a schema change cannot handled automatically, a warning with details will be logged.
+ - Mismatches in table schema from the Zongji binlog listener are now handled more gracefully.
+- Replication of wildcard tables is now supported.
+- Improved logging for binlog event processing.

modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts

@@ -162,9 +162,9 @@ export class MongoSyncBucketStorage
   async resolveTable(options: storage.ResolveTableOptions): Promise<storage.ResolveTableResult> {
     const { group_id, connection_id, connection_tag, entity_descriptor } = options;
 
-    const { schema, name: table, objectId, replicationColumns } = entity_descriptor;
+    const { schema, name, objectId, replicaIdColumns } = entity_descriptor;
 
-    const columns = replicationColumns.map((column) => ({
+    const normalizedReplicaIdColumns = replicaIdColumns.map((column) => ({
       name: column.name,
       type: column.type,
       type_oid: column.typeId
@@ -176,8 +176,8 @@ export class MongoSyncBucketStorage
         group_id: group_id,
         connection_id: connection_id,
         schema_name: schema,
-        table_name: table,
-        replica_id_columns2: columns
+        table_name: name,
+        replica_id_columns2: normalizedReplicaIdColumns
       };
       if (objectId != null) {
         filter.relation_id = objectId;
@@ -190,24 +190,24 @@ export class MongoSyncBucketStorage
           connection_id: connection_id,
           relation_id: objectId,
           schema_name: schema,
-          table_name: table,
+          table_name: name,
           replica_id_columns: null,
-          replica_id_columns2: columns,
+          replica_id_columns2: normalizedReplicaIdColumns,
           snapshot_done: false,
           snapshot_status: undefined
         };
 
         await col.insertOne(doc, { session });
       }
-      const sourceTable = new storage.SourceTable(
-        doc._id,
-        connection_tag,
-        objectId,
-        schema,
-        table,
-        replicationColumns,
-        doc.snapshot_done ?? true
-      );
+      const sourceTable = new storage.SourceTable({
+        id: doc._id,
+        connectionTag: connection_tag,
+        objectId: objectId,
+        schema: schema,
+        name: name,
+        replicaIdColumns: replicaIdColumns,
+        snapshotComplete: doc.snapshot_done ?? true
+      });
       sourceTable.syncEvent = options.sync_rules.tableTriggersEvent(sourceTable);
       sourceTable.syncData = options.sync_rules.tableSyncsData(sourceTable);
       sourceTable.syncParameters = options.sync_rules.tableSyncsParameters(sourceTable);
@@ -222,7 +222,7 @@ export class MongoSyncBucketStorage
 
       let dropTables: storage.SourceTable[] = [];
       // Detect tables that are either renamed, or have different replica_id_columns
-      let truncateFilter = [{ schema_name: schema, table_name: table }] as any[];
+      let truncateFilter = [{ schema_name: schema, table_name: name }] as any[];
       if (objectId != null) {
         // Only detect renames if the source uses relation ids.
         truncateFilter.push({ relation_id: objectId });
@@ -240,15 +240,16 @@ export class MongoSyncBucketStorage
         .toArray();
       dropTables = truncate.map(
         (doc) =>
-          new storage.SourceTable(
-            doc._id,
-            connection_tag,
-            doc.relation_id,
-            doc.schema_name,
-            doc.table_name,
-            doc.replica_id_columns2?.map((c) => ({ name: c.name, typeOid: c.type_oid, type: c.type })) ?? [],
-            doc.snapshot_done ?? true
-          )
+          new storage.SourceTable({
+            id: doc._id,
+            connectionTag: connection_tag,
+            objectId: doc.relation_id,
+            schema: doc.schema_name,
+            name: doc.table_name,
+            replicaIdColumns:
+              doc.replica_id_columns2?.map((c) => ({ name: c.name, typeOid: c.type_oid, type: c.type })) ?? [],
+            snapshotComplete: doc.snapshot_done ?? true
+          })
       );
 
       result = {
@@ -567,7 +568,6 @@ export class MongoSyncBucketStorage
             `${this.slot_name} Cleared batch of data in ${lib_mongo.db.MONGO_CLEAR_OPERATION_TIMEOUT_MS}ms, continuing...`
           );
           await timers.setTimeout(lib_mongo.db.MONGO_CLEAR_OPERATION_TIMEOUT_MS / 5);
-          continue;
         } else {
           throw e;
         }

modules/module-mongodb/src/api/MongoRouteAPIAdapter.ts

@@ -5,7 +5,7 @@ import * as sync_rules from '@powersync/service-sync-rules';
 import * as service_types from '@powersync/service-types';
 
 import { MongoManager } from '../replication/MongoManager.js';
-import { constructAfterRecord, createCheckpoint, STANDALONE_CHECKPOINT_ID } from '../replication/MongoRelation.js';
+import { constructAfterRecord, STANDALONE_CHECKPOINT_ID } from '../replication/MongoRelation.js';
 import { CHECKPOINTS_COLLECTION } from '../replication/replication-utils.js';
 import * as types from '../types/types.js';
 import { escapeRegExp } from '../utils.js';
@@ -137,15 +137,15 @@ export class MongoRouteAPIAdapter implements api.RouteAPI {
       if (tablePattern.isWildcard) {
         patternResult.tables = [];
         for (let collection of collections) {
-          const sourceTable = new SourceTable(
-            0,
-            this.connectionTag,
-            collection.name,
-            schema,
-            collection.name,
-            [],
-            true
-          );
+          const sourceTable = new SourceTable({
+            id: 0,
+            connectionTag: this.connectionTag,
+            objectId: collection.name,
+            schema: schema,
+            name: collection.name,
+            replicaIdColumns: [],
+            snapshotComplete: true
+          });
           let errors: service_types.ReplicationError[] = [];
           if (collection.type == 'view') {
             errors.push({ level: 'warning', message: `Collection ${schema}.${tablePattern.name} is a view` });
@@ -164,15 +164,15 @@ export class MongoRouteAPIAdapter implements api.RouteAPI {
           });
         }
       } else {
-        const sourceTable = new SourceTable(
-          0,
-          this.connectionTag,
-          tablePattern.name,
-          schema,
-          tablePattern.name,
-          [],
-          true
-        );
+        const sourceTable = new SourceTable({
+          id: 0,
+          connectionTag: this.connectionTag,
+          objectId: tablePattern.name,
+          schema: schema,
+          name: tablePattern.name,
+          replicaIdColumns: [],
+          snapshotComplete: true
+        });
 
         const syncData = sqlSyncRules.tableSyncsData(sourceTable);
         const syncParameters = sqlSyncRules.tableSyncsParameters(sourceTable);

modules/module-mongodb/src/replication/ChangeStream.ts

@@ -215,7 +215,7 @@ export class ChangeStream {
 
   async estimatedCountNumber(table: storage.SourceTable): Promise<number> {
     const db = this.client.db(table.schema);
-    return await db.collection(table.table).estimatedDocumentCount();
+    return await db.collection(table.name).estimatedDocumentCount();
   }
 
   private async getSnapshotLsn(): Promise<string> {
@@ -432,7 +432,7 @@ export class ChangeStream {
     const totalEstimatedCount = await this.estimatedCountNumber(table);
     let at = table.snapshotStatus?.replicatedCount ?? 0;
     const db = this.client.db(table.schema);
-    const collection = db.collection(table.table);
+    const collection = db.collection(table.name);
     await using query = new ChunkedSnapshotQuery({
       collection,
       key: table.snapshotStatus?.lastKey,

modules/module-mongodb/src/replication/MongoRelation.ts

@@ -13,7 +13,7 @@ export function getMongoRelation(source: mongo.ChangeStreamNameSpace): storage.S
     schema: source.db,
     // Not relevant for MongoDB - we use db + coll name as the identifier
     objectId: undefined,
-    replicationColumns: [{ name: '_id' }]
+    replicaIdColumns: [{ name: '_id' }]
   } satisfies storage.SourceEntityDescriptor;
 }
 
@@ -22,7 +22,7 @@ export function getMongoRelation(source: mongo.ChangeStreamNameSpace): storage.S
  */
 export function getCacheIdentifier(source: storage.SourceEntityDescriptor | storage.SourceTable): string {
   if (source instanceof storage.SourceTable) {
-    return `${source.schema}.${source.table}`;
+    return `${source.schema}.${source.name}`;
   }
   return `${source.schema}.${source.name}`;
 }

modules/module-mysql/package.json

@@ -33,9 +33,10 @@
     "@powersync/service-sync-rules": "workspace:*",
     "@powersync/service-types": "workspace:*",
     "@powersync/service-jsonbig": "workspace:*",
-    "@powersync/mysql-zongji": "0.2.0",
+    "@powersync/mysql-zongji": "^0.4.0",
     "async": "^3.2.4",
     "mysql2": "^3.11.0",
+    "node-sql-parser": "^5.3.9",
     "semver": "^7.5.4",
     "ts-codec": "^1.3.0",
     "uri-js": "^4.4.1",

modules/module-mysql/src/api/MySQLRouteAPIAdapter.ts

@@ -208,7 +208,7 @@ export class MySQLRouteAPIAdapter implements api.RouteAPI {
       idColumnsResult = await common.getReplicationIdentityColumns({
         connection: connection,
         schema,
-        table_name: tableName
+        tableName: tableName
       });
     } catch (ex) {
       idColumnsError = { level: 'fatal', message: ex.message };
@@ -217,7 +217,15 @@ export class MySQLRouteAPIAdapter implements api.RouteAPI {
     }
 
     const idColumns = idColumnsResult?.columns ?? [];
-    const sourceTable = new storage.SourceTable(0, this.config.tag, tableName, schema, tableName, idColumns, true);
+    const sourceTable = new storage.SourceTable({
+      id: 0,
+      connectionTag: this.config.tag,
+      objectId: tableName,
+      schema: schema,
+      name: tableName,
+      replicaIdColumns: idColumns,
+      snapshotComplete: true
+    });
     const syncData = syncRules.tableSyncsData(sourceTable);
     const syncParameters = syncRules.tableSyncsParameters(sourceTable);
 
@@ -232,7 +240,7 @@ export class MySQLRouteAPIAdapter implements api.RouteAPI {
     let selectError: service_types.ReplicationError | null = null;
     try {
       await this.retriedQuery({
-        query: `SELECT * FROM ${sourceTable.table} LIMIT 1`
+        query: `SELECT * FROM ${sourceTable.name} LIMIT 1`
       });
     } catch (e) {
       selectError = { level: 'fatal', message: e.message };

modules/module-mysql/src/common/common-index.ts

@@ -1,6 +1,5 @@
 export * from './check-source-configuration.js';
-export * from './get-replication-columns.js';
-export * from './get-tables-from-pattern.js';
+export * from './schema-utils.js';
 export * from './mysql-to-sqlite.js';
 export * from './read-executed-gtid.js';
 export * from './ReplicatedGTID.js';

modules/module-mysql/src/common/mysql-to-sqlite.ts

@@ -183,6 +183,9 @@ export function toSQLiteRow(row: Record<string, any>, columns: Map<string, Colum
           result[key] = row[key];
           break;
       }
+    } else {
+      // If the value is null, we just set it to null
+      result[key] = null;
     }
   }
   return sync_rules.toSyncRulesRow(result);