From e4b4aa818aab2b349b6a697857ea6d958b4b754b Mon Sep 17 00:00:00 2001
From: Jiri Pechanec <jpechane@redhat.com>
Date: Mon, 2 Mar 2020 10:14:43 +0100
Subject: [PATCH] DBZ-1834 Support for escaping

---
 .../mysql/antlr/MySqlAntlrDdlParser.java      | 12 +++++++--
 .../mysql/MySqlAntlrDdlParserTest.java        | 27 +++++++++++++++++++
 2 files changed, 37 insertions(+), 2 deletions(-)

diff --git a/debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/antlr/MySqlAntlrDdlParser.java b/debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/antlr/MySqlAntlrDdlParser.java
index 4f3dcaee0..c2fffb9b4 100644
--- a/debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/antlr/MySqlAntlrDdlParser.java
+++ b/debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/antlr/MySqlAntlrDdlParser.java
@@ -219,7 +219,8 @@ public TableId parseQualifiedTableId(MySqlParser.FullIdContext fullIdContext) {
         String tableName = null;
         final char EMPTY = '\0';
         char lastQuote = EMPTY;
-        for (char c : fullTableName) {
+        for (int i = 0; i < fullTableName.length; i++) {
+            char c = fullTableName[i];
             if (isQuote(c)) {
                 // Opening quote
                 if (lastQuote == EMPTY) {
@@ -227,7 +228,14 @@ public TableId parseQualifiedTableId(MySqlParser.FullIdContext fullIdContext) {
                 }
                 // Closing quote
                 else if (lastQuote == c) {
-                    lastQuote = EMPTY;
+                    // escape of quote by doubling
+                    if (i < fullTableName.length - 1 && fullTableName[i + 1] == c) {
+                        component.append(c);
+                        i++;
+                    }
+                    else {
+                        lastQuote = EMPTY;
+                    }
                 }
                 // Quote that is part of name
                 else {
diff --git a/debezium-connector-mysql/src/test/java/io/debezium/connector/mysql/MySqlAntlrDdlParserTest.java b/debezium-connector-mysql/src/test/java/io/debezium/connector/mysql/MySqlAntlrDdlParserTest.java
index e27ea671d..977272b3d 100644
--- a/debezium-connector-mysql/src/test/java/io/debezium/connector/mysql/MySqlAntlrDdlParserTest.java
+++ b/debezium-connector-mysql/src/test/java/io/debezium/connector/mysql/MySqlAntlrDdlParserTest.java
@@ -58,6 +58,33 @@ public void beforeEach() {
         tables = new Tables();
     }
 
+    @Test
+    @FixFor("DBZ-1834")
+    public void shouldHandleQuotes() {
+        String ddl = "CREATE TABLE mytable1 (`i.d` INT PRIMARY KEY)"
+                + "CREATE TABLE `mytable2` (`i.d` INT PRIMARY KEY)"
+                + "CREATE TABLE db.`mytable3` (`i.d` INT PRIMARY KEY)"
+                + "CREATE TABLE `db`.`mytable4` (`i.d` INT PRIMARY KEY)"
+                + "CREATE TABLE `db.mytable5` (`i.d` INT PRIMARY KEY)"
+                + "CREATE TABLE `db`.`myta``ble6` (`i.d` INT PRIMARY KEY)"
+                + "CREATE TABLE `db`.`mytable7``` (`i.d` INT PRIMARY KEY)"
+                + "CREATE TABLE ```db`.`mytable8` (`i.d` INT PRIMARY KEY)"
+                + "CREATE TABLE ```db`.`myta\"\"ble9` (`i.d` INT PRIMARY KEY)";
+        parser.parse(ddl, tables);
+        assertThat(((MySqlAntlrDdlParser) parser).getParsingExceptionsFromWalker().size()).isEqualTo(0);
+        assertThat(tables.size()).isEqualTo(9);
+
+        Assertions.assertThat(tables.forTable(null, null, "mytable1")).isNotNull();
+        Assertions.assertThat(tables.forTable(null, null, "mytable2")).isNotNull();
+        Assertions.assertThat(tables.forTable("db", null, "mytable3")).isNotNull();
+        Assertions.assertThat(tables.forTable("db", null, "mytable4")).isNotNull();
+        Assertions.assertThat(tables.forTable("db", null, "mytable5")).isNotNull();
+        Assertions.assertThat(tables.forTable("db", null, "myta`ble6")).isNotNull();
+        Assertions.assertThat(tables.forTable("db", null, "mytable7`")).isNotNull();
+        Assertions.assertThat(tables.forTable("`db", null, "mytable8")).isNotNull();
+        Assertions.assertThat(tables.forTable("`db", null, "myta\"\"ble9")).isNotNull();
+    }
+
     @Test
     @FixFor("DBZ-1645")
     public void shouldUpdateAndRenameTable() {