Merge pull request #64 from dukefirehawk/fix-bug/orm-mysql

Fix bug/orm mysql

packages/orm/angel_migration_runner/CHANGELOG.md

@@ -1,5 +1,10 @@
 # Change Log
 
+## 6.0.1
+
+* Added `MariaDbMigrationRunner` to support MariaDB migration with `mysql1` driver
+* Updated `MySqlMigrationRunner` to support MySQL migration with `mysql_client` driver
+
 ## 6.0.0
 
 * Updated to SDK 2.16.x

packages/orm/angel_migration_runner/README.md

@@ -10,5 +10,15 @@ Command-line based database migration runner for Angel3 ORM.
 Supported database:
 
 * PostgreSQL version 10 or later
-* MariaDB 10.2.x
-* MySQL 8.x
+* MariaDB 10.2.x or later
+* MySQL 8.x or later
+
+## Usage
+
+* For PostgreSQL, use `PostgresMigrationRunner` to perform the database migration.
+
+* For MariaDB, use `MariaDbMigrationRunner` to perform the database migration.
+
+* For MySQL, use `MySqlMigrationRunner` to perform the database migration.
+
+**Important Notes** For MariaDB and MySQL, both migration runner are using different drivers. MariaDB is using `mysql1` driver while MySQL is using `mysql_client` driver. This is necessary as neither driver works correctly over both MariaDB and MySQL. Based on testing, `mysql1` driver works seamlessly with MariaDB 10.2.x while `mysql_client` works well with MySQL 8.x.

packages/orm/angel_migration_runner/example/main.dart

@@ -27,7 +27,7 @@ void main(List<String> args) async {
       password: "Test123*",
       secure: false);
 
-  var mysqlMigrationRunner = MysqlMigrationRunner(
+  var mysqlMigrationRunner = MySqlMigrationRunner(
     mySQLConn,
     migrations: [
       UserMigration(),

packages/orm/angel_migration_runner/lib/mariadb.dart

@@ -0,0 +1,3 @@
+export 'src/mariadb/runner.dart';
+export 'src/mariadb/schema.dart';
+export 'src/mariadb/table.dart';

packages/orm/angel_migration_runner/lib/src/mariadb/runner.dart

@@ -0,0 +1,154 @@
+import 'dart:async';
+import 'dart:collection';
+import 'package:angel3_migration/angel3_migration.dart';
+import 'package:logging/logging.dart';
+import 'package:mysql1/mysql1.dart';
+import '../runner.dart';
+import '../util.dart';
+import 'schema.dart';
+
+class MariaDbMigrationRunner implements MigrationRunner {
+  final _log = Logger('MariaDbMigrationRunner');
+
+  final Map<String, Migration> migrations = {};
+  final Queue<Migration> _migrationQueue = Queue();
+  final MySqlConnection connection;
+  bool _connected = false;
+
+  MariaDbMigrationRunner(this.connection,
+      {Iterable<Migration> migrations = const [], bool connected = false}) {
+    if (migrations.isNotEmpty == true) migrations.forEach(addMigration);
+    _connected = connected == true;
+  }
+
+  @override
+  void addMigration(Migration migration) {
+    _migrationQueue.addLast(migration);
+  }
+
+  Future _init() async {
+    while (_migrationQueue.isNotEmpty) {
+      var migration = _migrationQueue.removeFirst();
+      var path = await absoluteSourcePath(migration.runtimeType);
+      migrations.putIfAbsent(path.replaceAll('\\', '\\\\'), () => migration);
+    }
+
+    if (!_connected) {
+      _connected = true;
+    }
+
+    await connection.query('''
+    CREATE TABLE IF NOT EXISTS migrations (
+      id serial,
+      batch integer,
+      path varchar(255),
+      PRIMARY KEY(id)
+    );
+    ''').then((result) {
+      _log.info('Check and create "migrations" table');
+    }).catchError((e) {
+      _log.severe('Failed to create "migrations" table.', e);
+    });
+  }
+
+  @override
+  Future up() async {
+    await _init();
+    var r = await connection.query('SELECT path from migrations;');
+    var existing = r.expand((x) => x).cast<String>();
+    var toRun = <String>[];
+
+    migrations.forEach((k, v) {
+      if (!existing.contains(k)) toRun.add(k);
+    });
+
+    if (toRun.isNotEmpty) {
+      var r = await connection.query('SELECT MAX(batch) from migrations;');
+      var rTmp = r.toList();
+      var curBatch = int.tryParse(rTmp[0][0] ?? '0') as int;
+      var batch = curBatch + 1;
+
+      for (var k in toRun) {
+        var migration = migrations[k]!;
+        var schema = MariaDbSchema();
+        migration.up(schema);
+        _log.info('Added "$k" into "migrations" table.');
+
+        try {
+          await schema.run(connection).then((_) async {
+            var result = await connection.query(
+                "INSERT INTO MIGRATIONS (batch, path) VALUES ($batch, '$k')");
+
+            return result.affectedRows;
+          });
+        } catch (e) {
+          _log.severe('Failed to insert into "migrations" table.', e);
+        }
+      }
+    } else {
+      _log.warning('Nothing to add into "migrations" table.');
+    }
+  }
+
+  @override
+  Future rollback() async {
+    await _init();
+
+    var r = await connection.query('SELECT MAX(batch) from migrations;');
+    var rTmp = r.toList();
+    var curBatch = int.tryParse(rTmp[0][0] ?? 0) as int;
+
+    r = await connection
+        .query('SELECT path from migrations WHERE batch = $curBatch;');
+    var existing = r.expand((x) => x).cast<String>();
+    var toRun = <String>[];
+
+    migrations.forEach((k, v) {
+      if (existing.contains(k)) toRun.add(k);
+    });
+
+    if (toRun.isNotEmpty) {
+      for (var k in toRun.reversed) {
+        var migration = migrations[k]!;
+        var schema = MariaDbSchema();
+        migration.down(schema);
+        _log.info('Removed "$k" from "migrations" table.');
+        await schema.run(connection).then((_) {
+          return connection
+              .query('DELETE FROM migrations WHERE path = \'$k\';');
+        });
+      }
+    } else {
+      _log.warning('Nothing to remove from "migrations" table.');
+    }
+  }
+
+  @override
+  Future reset() async {
+    await _init();
+    var r = await connection
+        .query('SELECT path from migrations ORDER BY batch DESC;');
+    var existing = r.expand((x) => x).cast<String>();
+    var toRun = existing.where(migrations.containsKey).toList();
+
+    if (toRun.isNotEmpty) {
+      for (var k in toRun.reversed) {
+        var migration = migrations[k]!;
+        var schema = MariaDbSchema();
+        migration.down(schema);
+        _log.info('Removed "$k" from "migrations" table.');
+        await schema.run(connection).then((_) {
+          return connection
+              .query('DELETE FROM migrations WHERE path = \'$k\';');
+        });
+      }
+    } else {
+      _log.warning('Nothing to remove from "migrations" table.');
+    }
+  }
+
+  @override
+  Future close() {
+    return connection.close();
+  }
+}

packages/orm/angel_migration_runner/lib/src/mariadb/schema.dart

@@ -0,0 +1,76 @@
+import 'dart:async';
+import 'package:angel3_migration/angel3_migration.dart';
+import 'package:logging/logging.dart';
+import 'package:mysql1/mysql1.dart';
+
+import 'table.dart';
+
+class MariaDbSchema extends Schema {
+  final _log = Logger('MariaDbSchema');
+
+  final int _indent;
+  final StringBuffer _buf;
+
+  MariaDbSchema._(this._buf, this._indent);
+
+  factory MariaDbSchema() => MariaDbSchema._(StringBuffer(), 0);
+
+  Future<int> run(MySqlConnection connection) async {
+    int affectedRows = 0;
+    await connection.transaction((ctx) async {
+      var sql = compile();
+      Results? result = await ctx.query(sql).catchError((e) {
+        _log.severe('Failed to run query: [ $sql ]', e);
+        throw e;
+      });
+      affectedRows = result?.affectedRows ?? 0;
+    });
+
+    return affectedRows;
+  }
+
+  String compile() => _buf.toString();
+
+  void _writeln(String str) {
+    for (var i = 0; i < _indent; i++) {
+      _buf.write('  ');
+    }
+
+    _buf.writeln(str);
+  }
+
+  @override
+  void drop(String tableName, {bool cascade = false}) {
+    var c = cascade == true ? ' CASCADE' : '';
+    _writeln('DROP TABLE "$tableName"$c;');
+  }
+
+  @override
+  void alter(String tableName, void Function(MutableTable table) callback) {
+    var tbl = MariaDbAlterTable(tableName);
+    callback(tbl);
+    _writeln('ALTER TABLE $tableName');
+    tbl.compile(_buf, _indent + 1);
+    _buf.write(';');
+  }
+
+  void _create(
+      String tableName, void Function(Table table) callback, bool ifNotExists) {
+    var op = ifNotExists ? ' IF NOT EXISTS' : '';
+    var tbl = MariaDbTable();
+    callback(tbl);
+    _writeln('CREATE TABLE$op $tableName (');
+    tbl.compile(_buf, _indent + 1);
+    _buf.writeln();
+    _writeln(');');
+  }
+
+  @override
+  void create(String tableName, void Function(Table table) callback) =>
+      _create(tableName, callback, false);
+
+  @override
+  void createIfNotExists(
+          String tableName, void Function(Table table) callback) =>
+      _create(tableName, callback, true);
+}

packages/orm/angel_migration_runner/lib/src/mariadb/table.dart

@@ -0,0 +1,168 @@
+import 'dart:collection';
+import 'package:angel3_orm/angel3_orm.dart';
+import 'package:angel3_migration/angel3_migration.dart';
+import 'package:charcode/ascii.dart';
+
+abstract class MariaDbGenerator {
+  static String columnType(MigrationColumn column) {
+    var str = column.type.name;
+    if (column.type.hasSize) {
+      return '$str(${column.length})';
+    } else {
+      return str;
+    }
+  }
+
+  static String compileColumn(MigrationColumn column) {
+    var buf = StringBuffer(columnType(column));
+
+    if (column.isNullable == false) buf.write(' NOT NULL');
+    if (column.defaultValue != null) {
+      String s;
+      var value = column.defaultValue;
+      if (value is RawSql) {
+        s = value.value;
+      } else if (value is String) {
+        var b = StringBuffer();
+        for (var ch in value.codeUnits) {
+          if (ch == $single_quote) {
+            b.write("\\'");
+          } else {
+            b.writeCharCode(ch);
+          }
+        }
+        s = b.toString();
+      } else {
+        s = value.toString();
+      }
+
+      buf.write(' DEFAULT $s');
+    }
+
+    if (column.indexType == IndexType.unique) {
+      buf.write(' UNIQUE');
+    } else if (column.indexType == IndexType.primaryKey) {
+      buf.write(' PRIMARY KEY');
+    }
+
+    for (var ref in column.externalReferences) {
+      buf.write(' ' + compileReference(ref));
+    }
+
+    return buf.toString();
+  }
+
+  static String compileReference(MigrationColumnReference ref) {
+    var buf = StringBuffer('REFERENCES ${ref.foreignTable}(${ref.foreignKey})');
+    if (ref.behavior != null) buf.write(' ' + ref.behavior!);
+    return buf.toString();
+  }
+}
+
+class MariaDbTable extends Table {
+  final Map<String, MigrationColumn> _columns = {};
+
+  @override
+  MigrationColumn declareColumn(String name, Column column) {
+    if (_columns.containsKey(name)) {
+      throw StateError('Cannot redeclare column "$name".');
+    }
+    var col = MigrationColumn.from(column);
+    _columns[name] = col;
+    return col;
+  }
+
+  void compile(StringBuffer buf, int indent) {
+    var i = 0;
+
+    _columns.forEach((name, column) {
+      var col = MariaDbGenerator.compileColumn(column);
+      if (i++ > 0) buf.writeln(',');
+
+      for (var i = 0; i < indent; i++) {
+        buf.write('  ');
+      }
+
+      buf.write('$name $col');
+    });
+  }
+}
+
+class MariaDbAlterTable extends Table implements MutableTable {
+  final Map<String, MigrationColumn> _columns = {};
+  final String tableName;
+  final Queue<String> _stack = Queue<String>();
+
+  MariaDbAlterTable(this.tableName);
+
+  void compile(StringBuffer buf, int indent) {
+    var i = 0;
+
+    while (_stack.isNotEmpty) {
+      var str = _stack.removeFirst();
+
+      if (i++ > 0) buf.writeln(',');
+
+      for (var i = 0; i < indent; i++) {
+        buf.write('  ');
+      }
+
+      buf.write(str);
+    }
+
+    if (i > 0) buf.writeln(';');
+
+    i = 0;
+    _columns.forEach((name, column) {
+      var col = MariaDbGenerator.compileColumn(column);
+      if (i++ > 0) buf.writeln(',');
+
+      for (var i = 0; i < indent; i++) {
+        buf.write('  ');
+      }
+
+      buf.write('ADD COLUMN $name $col');
+    });
+  }
+
+  @override
+  MigrationColumn declareColumn(String name, Column column) {
+    if (_columns.containsKey(name)) {
+      throw StateError('Cannot redeclare column "$name".');
+    }
+    var col = MigrationColumn.from(column);
+    _columns[name] = col;
+    return col;
+  }
+
+  @override
+  void dropNotNull(String name) {
+    _stack.add('ALTER COLUMN $name DROP NOT NULL');
+  }
+
+  @override
+  void setNotNull(String name) {
+    _stack.add('ALTER COLUMN $name SET NOT NULL');
+  }
+
+  @override
+  void changeColumnType(String name, ColumnType type, {int length = 256}) {
+    _stack.add('ALTER COLUMN $name TYPE ' +
+        MariaDbGenerator.columnType(MigrationColumn(type, length: length)));
+  }
+
+  @override
+  void renameColumn(String name, String newName) {
+    _stack.add('RENAME COLUMN $name TO "$newName"');
+  }
+
+  @override
+  void dropColumn(String name) {
+    _stack.add('DROP COLUMN $name');
+  }
+
+  @override
+  void rename(String newName) {
+    _stack.add('RENAME TO $newName');
+  }
+}

packages/orm/angel_migration_runner/lib/src/mysql/runner.dart

@@ -7,15 +7,15 @@ import '../runner.dart';
 import '../util.dart';
 import 'schema.dart';
 
-class MysqlMigrationRunner implements MigrationRunner {
-  final _log = Logger('PostgresMigrationRunner');
+class MySqlMigrationRunner implements MigrationRunner {
+  final _log = Logger('MysqlMigrationRunner');
 
   final Map<String, Migration> migrations = {};
   final Queue<Migration> _migrationQueue = Queue();
   final MySQLConnection connection;
   bool _connected = false;
 
-  MysqlMigrationRunner(this.connection,
+  MySqlMigrationRunner(this.connection,
       {Iterable<Migration> migrations = const [], bool connected = false}) {
     if (migrations.isNotEmpty == true) migrations.forEach(addMigration);
     _connected = connected == true;
@@ -34,16 +34,14 @@ class MysqlMigrationRunner implements MigrationRunner {
     }
 
     if (!_connected) {
-      //connection = await MySQLConnection.connect(settings);
-      await connection.connect();
       _connected = true;
     }
 
     await connection.execute('''
-    CREATE TABLE IF NOT EXISTS "migrations" (
+    CREATE TABLE IF NOT EXISTS migrations (
       id serial,
       batch integer,
-      path varchar,
+      path varchar(255),
       PRIMARY KEY(id)
     );
     ''').then((result) {
@@ -73,21 +71,22 @@ class MysqlMigrationRunner implements MigrationRunner {
 
       for (var k in toRun) {
         var migration = migrations[k]!;
-        var schema = MysqlSchema();
+        var schema = MySqlSchema();
         migration.up(schema);
         _log.info('Added "$k" into "migrations" table.');
-        await schema.run(connection).then((_) {
-          return connection.transactional((ctx) async {
-            var result = await ctx.execute(
-                "INSERT INTO MIGRATIONS (batch, path) VALUES ($batch, '$k')");
+        await schema.run(connection).then((_) async {
+          var result = await connection
+              .execute(
+                  "INSERT INTO MIGRATIONS (batch, path) VALUES ($batch, '$k')")
+              .catchError((e) {
+            _log.severe('Failed to insert into "migrations" table.', e);
+          });
 
-            return result.affectedRows;
+          return result.affectedRows.toInt();
         });
         //return connection.execute(
         //    'INSERT INTO MIGRATIONS (batch, path) VALUES ($batch, \'$k\');');
-        }).catchError((e) {
-          _log.severe('Failed to insert into "migrations" table.');
-        });
+
       }
     } else {
       _log.warning('Nothing to add into "migrations" table.');
@@ -115,7 +114,7 @@ class MysqlMigrationRunner implements MigrationRunner {
     if (toRun.isNotEmpty) {
       for (var k in toRun.reversed) {
         var migration = migrations[k]!;
-        var schema = MysqlSchema();
+        var schema = MySqlSchema();
         migration.down(schema);
         _log.info('Removed "$k" from "migrations" table.');
         await schema.run(connection).then((_) {
@@ -139,7 +138,7 @@ class MysqlMigrationRunner implements MigrationRunner {
     if (toRun.isNotEmpty) {
       for (var k in toRun.reversed) {
         var migration = migrations[k]!;
-        var schema = MysqlSchema();
+        var schema = MySqlSchema();
         migration.down(schema);
         _log.info('Removed "$k" from "migrations" table.');
         await schema.run(connection).then((_) {

packages/orm/angel_migration_runner/lib/src/mysql/schema.dart

@@ -1,30 +1,34 @@
 import 'dart:async';
 import 'package:angel3_migration/angel3_migration.dart';
-import 'package:angel3_migration_runner/src/mysql/table.dart';
 import 'package:logging/logging.dart';
 import 'package:mysql_client/mysql_client.dart';
 
-class MysqlSchema extends Schema {
+import 'table.dart';
+
+class MySqlSchema extends Schema {
   final _log = Logger('MysqlSchema');
 
   final int _indent;
   final StringBuffer _buf;
 
-  MysqlSchema._(this._buf, this._indent);
+  MySqlSchema._(this._buf, this._indent);
 
-  factory MysqlSchema() => MysqlSchema._(StringBuffer(), 0);
+  factory MySqlSchema() => MySqlSchema._(StringBuffer(), 0);
 
   Future<int> run(MySQLConnection connection) async {
     //return connection.execute(compile());
-    var result = await connection.transactional((ctx) async {
+    int affectedRows = 0;
+    await connection.transactional((ctx) async {
       var sql = compile();
       var result = await ctx.execute(sql).catchError((e) {
         _log.severe('Failed to run query: [ $sql ]', e);
       });
-      return result.affectedRows.toInt();
+      affectedRows = result.affectedRows.toInt();
+    }).catchError((e) {
+      _log.severe('Failed to run query in a transaction', e);
     });
 
-    return result;
+    return affectedRows;
   }
 
   String compile() => _buf.toString();
@@ -40,14 +44,14 @@ class MysqlSchema extends Schema {
   @override
   void drop(String tableName, {bool cascade = false}) {
     var c = cascade == true ? ' CASCADE' : '';
-    _writeln('DROP TABLE "$tableName"$c;');
+    _writeln('DROP TABLE $tableName$c;');
   }
 
   @override
   void alter(String tableName, void Function(MutableTable table) callback) {
     var tbl = MysqlAlterTable(tableName);
     callback(tbl);
-    _writeln('ALTER TABLE "$tableName"');
+    _writeln('ALTER TABLE $tableName');
     tbl.compile(_buf, _indent + 1);
     _buf.write(';');
   }
@@ -57,7 +61,7 @@ class MysqlSchema extends Schema {
     var op = ifNotExists ? ' IF NOT EXISTS' : '';
     var tbl = MysqlTable();
     callback(tbl);
-    _writeln('CREATE TABLE$op "$tableName" (');
+    _writeln('CREATE TABLE$op $tableName (');
     tbl.compile(_buf, _indent + 1);
     _buf.writeln();
     _writeln(');');

packages/orm/angel_migration_runner/lib/src/mysql/table.dart

@@ -3,7 +3,7 @@ import 'package:angel3_orm/angel3_orm.dart';
 import 'package:angel3_migration/angel3_migration.dart';
 import 'package:charcode/ascii.dart';
 
-abstract class MysqlGenerator {
+abstract class MySqlGenerator {
   static String columnType(MigrationColumn column) {
     var str = column.type.name;
     if (column.type.hasSize) {
@@ -53,8 +53,7 @@ abstract class MysqlGenerator {
   }
 
   static String compileReference(MigrationColumnReference ref) {
-    var buf =
-        StringBuffer('REFERENCES "${ref.foreignTable}"("${ref.foreignKey}")');
+    var buf = StringBuffer('REFERENCES ${ref.foreignTable}(${ref.foreignKey})');
     if (ref.behavior != null) buf.write(' ' + ref.behavior!);
     return buf.toString();
   }
@@ -77,14 +76,14 @@ class MysqlTable extends Table {
     var i = 0;
 
     _columns.forEach((name, column) {
-      var col = MysqlGenerator.compileColumn(column);
+      var col = MySqlGenerator.compileColumn(column);
       if (i++ > 0) buf.writeln(',');
 
       for (var i = 0; i < indent; i++) {
         buf.write('  ');
       }
 
-      buf.write('"$name" $col');
+      buf.write('$name $col');
     });
   }
 }
@@ -115,21 +114,21 @@ class MysqlAlterTable extends Table implements MutableTable {
 
     i = 0;
     _columns.forEach((name, column) {
-      var col = MysqlGenerator.compileColumn(column);
+      var col = MySqlGenerator.compileColumn(column);
       if (i++ > 0) buf.writeln(',');
 
       for (var i = 0; i < indent; i++) {
         buf.write('  ');
       }
 
-      buf.write('ADD COLUMN "$name" $col');
+      buf.write('ADD COLUMN $name $col');
     });
   }
 
   @override
   MigrationColumn declareColumn(String name, Column column) {
     if (_columns.containsKey(name)) {
-      throw StateError('Cannot redeclare column "$name".');
+      throw StateError('Cannot redeclare column $name.');
     }
     var col = MigrationColumn.from(column);
     _columns[name] = col;
@@ -138,32 +137,32 @@ class MysqlAlterTable extends Table implements MutableTable {
 
   @override
   void dropNotNull(String name) {
-    _stack.add('ALTER COLUMN "$name" DROP NOT NULL');
+    _stack.add('ALTER COLUMN $name DROP NOT NULL');
   }
 
   @override
   void setNotNull(String name) {
-    _stack.add('ALTER COLUMN "$name" SET NOT NULL');
+    _stack.add('ALTER COLUMN $name SET NOT NULL');
   }
 
   @override
   void changeColumnType(String name, ColumnType type, {int length = 256}) {
-    _stack.add('ALTER COLUMN "$name" TYPE ' +
-        MysqlGenerator.columnType(MigrationColumn(type, length: length)));
+    _stack.add('ALTER COLUMN $name TYPE ' +
+        MySqlGenerator.columnType(MigrationColumn(type, length: length)));
   }
 
   @override
   void renameColumn(String name, String newName) {
-    _stack.add('RENAME COLUMN "$name" TO "$newName"');
+    _stack.add('RENAME COLUMN $name TO $newName');
   }
 
   @override
   void dropColumn(String name) {
-    _stack.add('DROP COLUMN "$name"');
+    _stack.add('DROP COLUMN $name');
   }
 
   @override
   void rename(String newName) {
-    _stack.add('RENAME TO "$newName"');
+    _stack.add('RENAME TO $newName');
   }
 }

packages/orm/angel_migration_runner/lib/src/postgres/schema.dart

@@ -1,8 +1,8 @@
 import 'dart:async';
 import 'package:angel3_migration/angel3_migration.dart';
 import 'package:postgres/postgres.dart';
-import 'package:angel3_migration_runner/src/postgres/table.dart';
 import 'package:logging/logging.dart';
+import 'table.dart';
 
 class PostgresSchema extends Schema {
   final _log = Logger('PostgresSchema');

packages/orm/angel_migration_runner/pubspec.yaml

@@ -1,5 +1,5 @@
 name: angel3_migration_runner
-version: 6.0.0
+version: 6.0.1
 description: Command-line based database migration runner for Angel3's ORM.
 homepage: https://angel3-framework.web.app/
 repository: https://github.com/dukefirehawk/angel/tree/master/packages/orm/angel_migration_runner
@@ -11,7 +11,7 @@ dependencies:
   args: ^2.1.0
   charcode: ^1.2.0
   postgres: ^2.4.0
-  mysql_client: ^0.0.11
+  mysql_client: ^0.0.15
   mysql1: ^0.19.0
   logging: ^1.0.0
 dev_dependencies: