Fixed renaming of is_deleted column when the source columns have backticks

sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/MySqlDDLParserListenerImpl.java

@@ -109,8 +109,17 @@ public void enterColumnCreateTable(MySqlParser.ColumnCreateTableContext columnCr
         Set<String> columnNames = parseCreateTable(columnCreateTableContext, orderByColumns, partitionByColumn);
         //this.query.append(" Engine=")
         String isDeletedColumn = IS_DELETED_COLUMN;
-        if(columnNames.contains(isDeletedColumn)) {
-            isDeletedColumn = "__" + IS_DELETED_COLUMN;
+        // Iterate through columnNames and match isDeletedColumn with elements in columnNames.
+        // remove the backticks from elements in columnNames.
+        for(String columnName: columnNames) {
+            if(columnName.contains("`")) {
+               // replace backticks with empty string.
+                columnName = columnName.replace("`", "");
+            }
+            if(columnName.equalsIgnoreCase(isDeletedColumn)) {
+                isDeletedColumn = "__" + IS_DELETED_COLUMN;
+                break;
+            }
         }
 
         // Check if the destination is ReplicatedReplacingMergeTree.

sink-connector-lightweight/src/test/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/MySqlDDLParserListenerImplTest.java

@@ -777,6 +777,33 @@ public void testAlterDatabaseAddColumnJson() {
         Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase(clickhouseExpectedQuery));
     }
 
+    @Test
+    public void testRenameIsDeletedColumn() {
+        String sql = "CREATE TABLE `city` (\n" +
+                "  `ID` int NOT NULL AUTO_INCREMENT,\n" +
+                "  `Name` char(35) COLLATE utf8mb4_general_ci NOT NULL DEFAULT '',\n" +
+                "  `CountryCode` char(3) COLLATE utf8mb4_general_ci NOT NULL DEFAULT '',\n" +
+                "  `District` char(20) COLLATE utf8mb4_general_ci NOT NULL DEFAULT '',\n" +
+                "  `Population` int NOT NULL DEFAULT '0',\n" +
+                "  `is_deleted` tinyint(1) DEFAULT '0',\n" +
+                "  PRIMARY KEY (`ID`)\n" +
+                ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;";
+
+        StringBuffer clickHouseQuery = new StringBuffer();
+        mySQLDDLParserService.parseSql(sql, "employees", clickHouseQuery);
+
+        Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase(
+                "CREATE TABLE employees.`city`(`ID` Int32 NOT NULL ,`Name` String NOT NULL ,`CountryCode` String NOT NULL ,`District` String NOT NULL ,`Population` Int32 NOT NULL ,`is_deleted` Nullable(Int16),`_version` UInt64,`__is_deleted` UInt8) Engine=ReplacingMergeTree(_version,__is_deleted) ORDER BY (`ID`)"));
+
+
+        String sqlWithoutBackticks = "create table city(id int not null auto_increment, Name char(35) , is_deleted tinyint(1) DEFAULT 0, primary key(id))";
+
+        StringBuffer clickHouseQuery2 = new StringBuffer();
+        mySQLDDLParserService.parseSql(sqlWithoutBackticks, "employees", clickHouseQuery2);
+
+        Assert.assertTrue(clickHouseQuery2.toString().equalsIgnoreCase(
+                "CREATE TABLE employees.city(id Int32 NOT NULL ,Name Nullable(String),is_deleted Nullable(Int16),`_version` UInt64,`__is_deleted` UInt8) Engine=ReplacingMergeTree(_version,__is_deleted) ORDER BY (id)"));
+    }
 //    @Test
 //    public void deleteData() {
 //        String sql = "DELETE FROM Customers WHERE CustomerName='Alfreds Futterkiste'";