alexey.rassokhin/tet123
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

DBZ-57 Added support for the shortened CHARSET alias for CHARACTER SET in MySQL DDL statements

Browse Source 
Added explicit support for handling `CHARSET` as an alias for `CHARACTER SET` in both tables and columns.  `CREATE DATABASE` and `ALTER DATABASE` statements can also specify character sets, but the DDL parser handles but does not explicitly parse them so no modification is needed for them. Several unit tests were added to confirm the behavior.
This commit is contained in:
Randall Hauch 2016-05-19 18:11:11 -05:00
parent 4f40cc8332
commit c20b49a8fc
2 changed files with 99 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/MySqlDdlParser.java
View File

@ -80,6 +80,10 @@ protected void initializeDataTypes(DataTypeParser dataTypes) {
dataTypes.register(Types.BLOB, "VARCHAR(L) BINARY [CHARACTER SET charset_name] [COLLATE collation_name]");
dataTypes.register(Types.VARCHAR, "CHAR[(L)] [CHARACTER SET charset_name] [COLLATE collation_name]");
dataTypes.register(Types.VARCHAR, "VARCHAR(L) [CHARACTER SET charset_name] [COLLATE collation_name]");
dataTypes.register(Types.BLOB, "CHAR[(L)] BINARY [CHARSET charset_name] [COLLATE collation_name]");
dataTypes.register(Types.BLOB, "VARCHAR(L) BINARY [CHARSET charset_name] [COLLATE collation_name]");
dataTypes.register(Types.VARCHAR, "CHAR[(L)] [CHARSET charset_name] [COLLATE collation_name]");
dataTypes.register(Types.VARCHAR, "VARCHAR(L) [CHARSET charset_name] [COLLATE collation_name]");
dataTypes.register(Types.CHAR, "BINARY[(L)]");
dataTypes.register(Types.VARBINARY, "VARBINARY(L)");
dataTypes.register(Types.BLOB, "TINYBLOB");
@ -96,6 +100,16 @@ protected void initializeDataTypes(DataTypeParser dataTypes) {
dataTypes.register(Types.VARCHAR, "LONGTEXT [CHARACTER SET charset_name] [COLLATE collation_name]");
dataTypes.register(Types.CHAR, "ENUM(...) [CHARACTER SET charset_name] [COLLATE collation_name]");
dataTypes.register(Types.CHAR, "SET(...) [CHARACTER SET charset_name] [COLLATE collation_name]");
dataTypes.register(Types.BLOB, "TINYTEXT BINARY [CHARSET charset_name] [COLLATE collation_name]");
dataTypes.register(Types.BLOB, "TEXT BINARY [CHARSET charset_name] [COLLATE collation_name]");
dataTypes.register(Types.BLOB, "MEDIUMTEXT BINARY [CHARSET charset_name] [COLLATE collation_name]");
dataTypes.register(Types.BLOB, "LONGTEXT BINARY [CHARSET charset_name] [COLLATE collation_name]");
dataTypes.register(Types.VARCHAR, "TINYTEXT [CHARSET charset_name] [COLLATE collation_name]");
dataTypes.register(Types.VARCHAR, "TEXT [CHARSET charset_name] [COLLATE collation_name]");
dataTypes.register(Types.VARCHAR, "MEDIUMTEXT [CHARSET charset_name] [COLLATE collation_name]");
dataTypes.register(Types.VARCHAR, "LONGTEXT [CHARSET charset_name] [COLLATE collation_name]");
dataTypes.register(Types.CHAR, "ENUM(...) [CHARSET charset_name] [COLLATE collation_name]");
dataTypes.register(Types.CHAR, "SET(...) [CHARSET charset_name] [COLLATE collation_name]");
dataTypes.register(Types.OTHER, "JSON");
}
@ -839,12 +853,16 @@ protected void parseAlterSpecification(Marker start, TableEditor table, Consumer
newTableName.accept(newTableId);
} else if (tokens.canConsume("ORDER", "BY")) {
consumeCommaSeparatedValueList(start); // this should not affect the order of the columns in the table
} else if (tokens.canConsume("CONVERT", "TO", "CHARACTER", "SET")) {
} else if (tokens.canConsume("CONVERT", "TO", "CHARACTER", "SET")
|| tokens.canConsume("CONVERT", "TO", "CHARSET")) {
tokens.consume(); // charset name
if (tokens.canConsume("COLLATE")) {
tokens.consume(); // collation name
}
} else if (tokens.canConsume("CHARACTER", "SET") || tokens.canConsume("DEFAULT", "CHARACTER", "SET")) {
} else if (tokens.canConsume("CHARACTER", "SET")
|| tokens.canConsume("CHARSET")
|| tokens.canConsume("DEFAULT", "CHARACTER", "SET")
|| tokens.canConsume("DEFAULT", "CHARSET")) {
tokens.canConsume('=');
tokens.consume(); // charset name
if (tokens.canConsume("COLLATE")) {

79
debezium-connector-mysql/src/test/java/io/debezium/connector/mysql/MySqlDdlParserTest.java
View File

@ -138,6 +138,85 @@ public void shouldParseCreateTableStatementWithSignedTypes() {
assertColumn(foo, "c2", "INT UNSIGNED", Types.INTEGER, -1, -1, false, false, false);
}
@Test
public void shouldParseCreateTableStatementWithCharacterSetForTable() {
String ddl = "CREATE TABLE t ( col1 VARCHAR(25) ) DEFAULT CHARACTER SET utf8 DEFAULT COLLATE utf8_general_ci; ";
parser.parse(ddl, tables);
assertThat(tables.size()).isEqualTo(1);
Table t = tables.forTable(new TableId(null, null, "t"));
assertThat(t).isNotNull();
assertThat(t.columnNames()).containsExactly("col1");
assertThat(t.primaryKeyColumnNames()).isEmpty();
assertColumn(t, "col1", "VARCHAR", Types.VARCHAR, 25, -1, true, false, false);
ddl = "CREATE TABLE t2 ( col1 VARCHAR(25) ) DEFAULT CHARSET utf8 DEFAULT COLLATE utf8_general_ci; ";
parser.parse(ddl, tables);
assertThat(tables.size()).isEqualTo(2);
Table t2 = tables.forTable(new TableId(null, null, "t2"));
assertThat(t2).isNotNull();
assertThat(t2.columnNames()).containsExactly("col1");
assertThat(t2.primaryKeyColumnNames()).isEmpty();
assertColumn(t2, "col1", "VARCHAR", Types.VARCHAR, 25, -1, true, false, false);
ddl = "CREATE TABLE t3 ( col1 VARCHAR(25) ) CHARACTER SET utf8 COLLATE utf8_general_ci; ";
parser.parse(ddl, tables);
assertThat(tables.size()).isEqualTo(3);
Table t3 = tables.forTable(new TableId(null, null, "t3"));
assertThat(t3).isNotNull();
assertThat(t3.columnNames()).containsExactly("col1");
assertThat(t3.primaryKeyColumnNames()).isEmpty();
assertColumn(t3, "col1", "VARCHAR", Types.VARCHAR, 25, -1, true, false, false);
ddl = "CREATE TABLE t4 ( col1 VARCHAR(25) ) CHARSET utf8 COLLATE utf8_general_ci; ";
parser.parse(ddl, tables);
assertThat(tables.size()).isEqualTo(4);
Table t4 = tables.forTable(new TableId(null, null, "t4"));
assertThat(t4).isNotNull();
assertThat(t4.columnNames()).containsExactly("col1");
assertThat(t4.primaryKeyColumnNames()).isEmpty();
assertColumn(t4, "col1", "VARCHAR", Types.VARCHAR, 25, -1, true, false, false);
}
@Test
public void shouldParseCreateTableStatementWithCharacterSetForColumns() {
String ddl = "CREATE TABLE t ( col1 VARCHAR(25) CHARACTER SET greek ); ";
parser.parse(ddl, tables);
assertThat(tables.size()).isEqualTo(1);
Table t = tables.forTable(new TableId(null, null, "t"));
assertThat(t).isNotNull();
assertThat(t.columnNames()).containsExactly("col1");
assertThat(t.primaryKeyColumnNames()).isEmpty();
assertColumn(t, "col1", "VARCHAR CHARACTER SET greek", Types.VARCHAR, 25, -1, true, false, false);
}
@Test
public void shouldParseAlterTableStatementThatAddsCharacterSetForColumns() {
String ddl = "CREATE TABLE t ( col1 VARCHAR(25) ); ";
parser.parse(ddl, tables);
assertThat(tables.size()).isEqualTo(1);
Table t = tables.forTable(new TableId(null, null, "t"));
assertThat(t).isNotNull();
assertThat(t.columnNames()).containsExactly("col1");
assertThat(t.primaryKeyColumnNames()).isEmpty();
assertColumn(t, "col1", "VARCHAR", Types.VARCHAR, 25, -1, true, false, false);
ddl = "ALTER TABLE t MODIFY col1 VARCHAR(50) CHARACTER SET greek;";
parser.parse(ddl, tables);
Table t2 = tables.forTable(new TableId(null, null, "t"));
assertThat(t2).isNotNull();
assertThat(t2.columnNames()).containsExactly("col1");
assertThat(t2.primaryKeyColumnNames()).isEmpty();
assertColumn(t2, "col1", "VARCHAR CHARACTER SET greek", Types.VARCHAR, 50, -1, true, false, false);
ddl = "ALTER TABLE t MODIFY col1 VARCHAR(75) CHARSET utf8;";
parser.parse(ddl, tables);
Table t3 = tables.forTable(new TableId(null, null, "t"));
assertThat(t3).isNotNull();
assertThat(t3.columnNames()).containsExactly("col1");
assertThat(t3.primaryKeyColumnNames()).isEmpty();
assertColumn(t3, "col1", "VARCHAR CHARSET utf8", Types.VARCHAR, 75, -1, true, false, false);
}
@Test
public void shouldParseGrantStatement() {
String ddl = "GRANT ALL PRIVILEGES ON `mysql`.* TO 'mysqluser'@'%'";