DBZ-49 MySQL DDL parser is more tolerant of REFERENCE clauses in CREA…

…TE TABLE statements

MySQL 5.6 using the MyISAM engine will create the `help_relation` system table using a CREATE TABLE statement that does not have in the columns' REFERENCE clause a list of columns in the referenced table. MySQL 5.7 using the InnoDB engine does not include the REFERENCE clauses.

Because Debezium's MySQL DDL parser is meant only to understand the statements recorded in the binlog, it does not have to validate the statements and therefore the DDL parser can be a bit more lenient by not requiring the list of columns in a REFERENCE clause in a CREATE TABLE statement's column definitions.

This commit also adds several unit tests that validate all of the DDL statements used by MySQL 5.6 and 5.7 during startup (in the configurations used in our integration tests).

debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/MySqlDdlParser.java

@@ -578,7 +578,9 @@ protected void parseIndexOptions(Marker start) {
     protected void parseReferenceDefinition(Marker start) {
         tokens.consume("REFERENCES");
         parseSchemaQualifiedName(start); // table name
-        parseColumnNameList(start);
+        if (tokens.matches('(')) {
+            parseColumnNameList(start);
+        }
         if (tokens.canConsume("MATCH")) {
             tokens.consumeAnyOf("FULL", "PARTIAL", "SIMPLE");
             if (tokens.canConsume("ON")) {

debezium-connector-mysql/src/test/java/io/debezium/connector/mysql/MySqlDdlParserTest.java

@@ -150,28 +150,44 @@ public void shouldParseGrantStatement() {
     public void shouldParseCreateStatements() {
         parser.parse(readFile("ddl/mysql-test-create.ddl"), tables);
         Testing.print(tables);
-        assertThat(tables.size()).isEqualTo(57); // no tables
+        assertThat(tables.size()).isEqualTo(57);
         assertThat(listener.total()).isEqualTo(144);
     }
 
     @Test
     public void shouldParseTestStatements() {
         parser.parse(readFile("ddl/mysql-test-statements.ddl"), tables);
         Testing.print(tables);
-        assertThat(tables.size()).isEqualTo(6); // no tables
+        assertThat(tables.size()).isEqualTo(6);
         assertThat(listener.total()).isEqualTo(49);
         //listener.forEach(this::printEvent);
     }
 
     @Test
     public void shouldParseSomeLinesFromCreateStatements() {
         parser.parse(readLines(189,"ddl/mysql-test-create.ddl"), tables);
-        assertThat(tables.size()).isEqualTo(39); // no tables
+        assertThat(tables.size()).isEqualTo(39);
         assertThat(listener.total()).isEqualTo(120);
     }
+
+    @Test
+    public void shouldParseMySql56InitializationStatements() {
+        parser.parse(readLines(1,"ddl/mysql-test-init-5.6.ddl"), tables);
+        assertThat(tables.size()).isEqualTo(85); // 1 table
+        assertThat(listener.total()).isEqualTo(112);
+        listener.forEach(this::printEvent);
+    }
+
+    @Test
+    public void shouldParseMySql57InitializationStatements() {
+        parser.parse(readLines(1,"ddl/mysql-test-init-5.7.ddl"), tables);
+        assertThat(tables.size()).isEqualTo(123);
+        assertThat(listener.total()).isEqualTo(125);
+        listener.forEach(this::printEvent);
+    }
 
     protected void printEvent( Event event ) {
-        System.out.println(event);
+        Testing.print(event);
     }
 
     protected String readFile( String classpathResource ) {

debezium-connector-mysql/src/test/java/io/debezium/connector/mysql/ReadBinLogIT.java

@@ -239,6 +239,29 @@ public void shouldCaptureMultipleWriteUpdateDeletesInSingleEvents() throws Excep
                                                  .removedRow("Jamie", 19, any(), any()));
     }
 
+    /**
+     * Test case that is normally commented out since it is only useful to print out the DDL statements recorded by
+     * the binlog during a MySQL server initialization and startup.
+     * 
+     * @throws Exception if there are problems
+     */
+    @Ignore
+    @Test
+    public void shouldCaptureQueryEventData() throws Exception {
+        // Testing.Print.enable();
+        startClient(client -> {
+            client.setBinlogFilename("mysql-bin.000001");
+            client.setBinlogPosition(4);
+        });
+        counters.consumeAll(5, TimeUnit.SECONDS);
+        List<QueryEventData> allQueryEvents = recordedEventData(QueryEventData.class, -1);
+        allQueryEvents.forEach(event -> {
+            String sql = event.getSql();
+            if (sql.equalsIgnoreCase("BEGIN") || sql.equalsIgnoreCase("COMMIT")) return;
+            System.out.println(event.getSql());
+        });
+    }
+
     @Test
     public void shouldQueryInformationSchema() throws Exception {
         // long tableId = writeRows.getTableId();