[MySQL] Automatic schema change handling (#287)

* Removed zongji type mappings which are now provided by the Zongji package directly
Added check for tablemap events

* Moved most of the binlog event handling logic to a separate BinlogListener class.
Introduced a mechanism to limit the maximum size of the binlog processing queue, thus also limiting memory usage.
This maximum processing queue size is configurable

* Updated the BinLogStream to use the new BinLogListener

* Renamed BinlogListener to BinLogListener

* Added changeset

* Simplified BinLogListener stopping mechanism
Cleaned up BinLogStream logs a bit

* Corrected BinLogListener name.
Simplified BinLogListener stopping mechanism

* Supply port for binlog listener connections.

* Only set up binlog heartbeat once the listener is fully started up.

Added a few more defensive stopped checks to the binlog listener

* Updated changeset

* Changed binlog backpressure mechanism to be based on processing queue memory usage rather than number of events

* Changed binlog backpressure mechanism to be based on processing queue memory usage rather than number of events.

Introduced a maximum timeout that the binlog processing queue can be paused before auto-resuming. This is to prevent the replication connection timing out.

* Added optional columns field to SourceEntityDescriptor
Made SourceTable implement SourceEntityDescriptor interface

* Cleanup unused imports

* Ensure column values are preserved when available Report 0 storage metrics instead of ignoring them. SourceTable.

Moved MySQL table detail retrieval logic to utility function.

* Added basic schema change handling for MySQL

* Revert columns field addition to SourceEntityDescriptor

* Added schema change handling for the MySQL binlog replication.

* Include powersync core version in metrics metadata

* Code analysis cleanup

* Merge conflicts

* Fixed parser import

* Fixed mysql->sqlite rows parsing that would filter out columns with null values

* Cleaned up SchemaChange handling in BinLogListener
Improved binlog table filtering
Added extended type definitions for node-sql-parser package

* Added schema change tests
Cleaned up MySQL tests in general and added a few new test utils

* Change binlog event receive log message to debug

* Revert and fix mysql->sqlite row conversion for null value columns

* Added conditional skip of mysql schema test for syntax that does not exist in version 5.7

* Fixed version checking for mysql 5.7 incompatible test

* Fix skip test on mysql 5.7 schema change

* Reverted mysql dev docker compose
Updated to released zongji listener version

* Moved schema change handling to processing queue
Catch parsing errors, and log an error if the DDL query might apply to one of the tables in the sync rules.

* Fixed bug where multiple zongji listeners could be started if multiple schema change events were in the processing queue
Added small timeout to test to prevent rare race condition

* Extended node-sql-parser type definitions
Added util functions to identify the different types of DDL statements

* - Simplified schema change types
- Added more detections of constraint changes
- Removed detection of create table statements since they can be detected and reacted to when row events are received for new tables
- Added multiple extra test cases

* Removed unused constant

* Skip unsupported schema test for MySQL 5.7

* Added error handling for zongji emitted schema errors

* Added changeset

* Typo fixes from pr feedback

* Removed filters from mysql dev docker config

* Added safeguard for gtid splitting when no transactions have been run on the mysql database yet.

* BinLog listener now correctly takes schema into account for replication.
TableFilter creation is now internally handled in the BinLog listener
Pause/unpause binlog listening now uses the same stop start functionality used for schema change handling.

* BinLog stream now correctly honors multiple schemas in the sync rules.

* Added tests for multi schema support

* MySQL util fix post merge

* Removed accidentally commited keepalive code in BinLogStream.

* Cleaned up Binlog docs and comments a bit

* Removed potentially spammy log entry.

Author: Rentacookie

.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

@@ -164,9 +164,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
@@ -178,8 +178,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;
@@ -192,24 +192,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);
@@ -224,7 +224,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 });
@@ -242,15 +242,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 = {
@@ -577,7 +578,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();
   }
 
   /**
@@ -449,7 +449,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/dev/docker/mysql/init-scripts/my.cnf

@@ -4,6 +4,4 @@ enforce-gtid-consistency = ON
 # Row format required for ZongJi
 binlog_format = row
 log_bin=mysql-bin
-server-id=1
-binlog-do-db=mydatabase
-replicate-do-table=mydatabase.lists
+server-id=1

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/ReplicatedGTID.ts

@@ -92,10 +92,15 @@ export class ReplicatedGTID {
    * @returns A comparable string in the format
    *   `padded_end_transaction|raw_gtid|binlog_filename|binlog_position`
    */
-  get comparable() {
+  get comparable(): string {
     const { raw, position } = this;
     const [, transactionRanges] = this.raw.split(':');
 
+    // This means no transactions have been executed on the database yet
+    if (!transactionRanges) {
+      return ReplicatedGTID.ZERO.comparable;
+    }
+
     let maxTransactionId = 0;
 
     for (const range of transactionRanges.split(',')) {

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';