[split 17/22] ddl parser: case-insensitive built-in ignore patterns, DDL translation regression suite

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

@@ -201,4 +201,20 @@ public class Constants {
      */
     public static final Set<String> NULLABLE_NOT_SUPPORTED_DATA_TYPES =
             new HashSet<>(Arrays.asList("point", "polygon"));
+
+    /**
+     * Backtick-escapes a SQL identifier (table name, column name, database name)
+     * to safely handle reserved words and special characters.
+     * Strips any existing backticks first to avoid double-escaping.
+     *
+     * @param identifier The identifier to escape.
+     * @return The backtick-escaped identifier, or null if input is null.
+     */
+    public static String escapeIdentifier(String identifier) {
+        if (identifier == null) {
+            return null;
+        }
+        String stripped = identifier.replace("`", "");
+        return "`" + stripped + "`";
+    }
 }

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

@@ -41,8 +41,9 @@ public void syntaxError(Recognizer<?, ?> recognizer,
                             Object offendingSymbol, int line,
                             int charPositionInLine, String msg,
                             RecognitionException e) {
-        log.error("Error parsing");
-        throw new RuntimeException("Error parsing DDL");
+        String errorDetail = String.format("Error parsing DDL at line %d:%d - %s", line, charPositionInLine, msg);
+        log.error(errorDetail);
+        throw new RuntimeException(errorDetail);
     }
 
     /**

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

@@ -11,6 +11,8 @@
 import org.antlr.v4.runtime.CommonTokenStream;
 import org.antlr.v4.runtime.Token;
 import org.antlr.v4.runtime.tree.ParseTreeWalker;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -27,6 +29,11 @@
 @Singleton
 public class MySQLDDLParserService implements DDLParserService {
 
+    /**
+     * Logger for the MySQLDDLParserService class.
+     */
+    private static final Logger log = LogManager.getLogger(MySQLDDLParserService.class);
+
     /**
      * The name of the database being processed.
      */
@@ -101,7 +108,7 @@ public String parseSql(String sql, String tableName, StringBuffer parsedQuery) {
         ParseTreeWalker walker = new ParseTreeWalker();
         walker.walk(listener, parser.root());
 
-        return clickHouseResult;
+        return parsedQuery.toString();
     }
 
     /**
@@ -134,7 +141,7 @@ public String parseSql(String sql, String tableName,  StringBuffer parsedQuery,
         // Set the drop or truncate flag
         isDropOrTruncate.set(isDropOrTruncateStatement(tokens));
 
-        return clickHouseResult;
+        return parsedQuery.toString();
     }
 
     /**
@@ -143,16 +150,152 @@ public String parseSql(String sql, String tableName,  StringBuffer parsedQuery,
      * @param tokens the list of tokens generated by the lexer.
      * @return true if the statement is DROP or TRUNCATE, false otherwise.
      */
+    /**
+     * Extracts the table name a DDL statement operates on.
+     *
+     * <p>Used to key DDL schema-cache invalidation. Debezium's SchemaChangeKey does
+     * not carry a table name, so it must be recovered from the DDL text itself.</p>
+     *
+     * <p>Returns the bare table name (schema qualifier and quoting removed), or null
+     * when the statement has no single table subject (e.g. CREATE/DROP DATABASE) or
+     * cannot be tokenised.</p>
+     *
+     * @param sql the DDL statement.
+     * @return the table name, or null if it cannot be determined.
+     */
+    public static String extractTableName(String sql) {
+        if (sql == null || sql.trim().isEmpty()) {
+            return null;
+        }
+        try {
+            MySqlLexer lexer = new MySqlLexer(
+                    new CaseChangingCharStream(CharStreams.fromString(sql), true));
+            CommonTokenStream tokens = new CommonTokenStream(lexer);
+            tokens.fill();
+
+            List<Token> tokensList = tokens.getTokens();
+            boolean sawTable = false;
+            for (Token token : tokensList) {
+                if (token.getChannel() != Token.DEFAULT_CHANNEL
+                        || token.getType() == Token.EOF) {
+                    continue;
+                }
+                int type = token.getType();
+                if (type == MySqlParser.TABLE) {
+                    sawTable = true;
+                    continue;
+                }
+                if (!sawTable) {
+                    continue;
+                }
+                // Skip the optional qualifiers that may follow TABLE.
+                if (type == MySqlParser.IF || type == MySqlParser.EXISTS
+                        || type == MySqlParser.NOT) {
+                    continue;
+                }
+                // The identifier may be schema-qualified. Two lexer shapes occur:
+                //   `db`.`tbl`  -> ID(`db`), DOT_ID(.`tbl`)  or three tokens
+                //   db.tbl      -> ID(db),   DOT_ID(.tbl)
+                // Keep consuming trailing dotted components and return the LAST
+                // one. Returning the first would yield the DATABASE name and
+                // invalidate the wrong cache key, leaving the real DbWriter stale.
+                String identifier = token.getText();
+                int index = tokensList.indexOf(token);
+                while (true) {
+                    Token next = nextDefaultChannelToken(tokensList, index);
+                    if (next == null) {
+                        break;
+                    }
+                    String nextText = next.getText();
+                    if (nextText != null && nextText.startsWith(".")
+                            && nextText.length() > 1) {
+                        // Single DOT_ID token, e.g. ".orders".
+                        identifier = nextText.substring(1);
+                        index = tokensList.indexOf(next);
+                        continue;
+                    }
+                    if (".".equals(nextText)) {
+                        // Separate dot token; the component follows it.
+                        Token after = nextDefaultChannelToken(
+                                tokensList, tokensList.indexOf(next));
+                        if (after == null) {
+                            break;
+                        }
+                        identifier = after.getText();
+                        index = tokensList.indexOf(after);
+                        continue;
+                    }
+                    break;
+                }
+                return normalizeTableName(identifier);
+            }
+        } catch (Exception e) {
+            log.debug("Unable to extract table name from DDL: {}", sql, e);
+        }
+        return null;
+    }
+
+    /**
+     * Returns the next token on the default channel after the given index, or null.
+     *
+     * @param tokensList the full token list.
+     * @param fromIndex  the index to search after.
+     * @return the next default-channel token, or null if there is none.
+     */
+    private static Token nextDefaultChannelToken(List<Token> tokensList, int fromIndex) {
+        for (int i = fromIndex + 1; i < tokensList.size(); i++) {
+            Token candidate = tokensList.get(i);
+            if (candidate.getChannel() != Token.DEFAULT_CHANNEL) {
+                continue;
+            }
+            if (candidate.getType() == Token.EOF) {
+                return null;
+            }
+            return candidate;
+        }
+        return null;
+    }
+
+    /**
+     * Strips backtick/quote characters and any database qualifier from an identifier.
+     *
+     * @param rawIdentifier the raw identifier token text.
+     * @return the bare table name, or null if empty.
+     */
+    private static String normalizeTableName(String rawIdentifier) {
+        if (rawIdentifier == null) {
+            return null;
+        }
+        String name = rawIdentifier.replace("`", "").replace("\"", "").trim();
+        int lastDot = name.lastIndexOf('.');
+        if (lastDot >= 0 && lastDot < name.length() - 1) {
+            name = name.substring(lastDot + 1);
+        }
+        return name.isEmpty() ? null : name;
+    }
+
     public boolean isDropOrTruncateStatement(CommonTokenStream tokens) {
 
-        boolean result = false;
         List<Token> tokensList = tokens.getTokens();
 
-        if (tokensList.stream().anyMatch(x -> x.getType() == MySqlParser.DROP || x.getType() == MySqlParser.TRUNCATE)) {
-            result = true;
+        // Only the LEADING keyword decides whether this statement destroys data.
+        // Scanning the whole token stream for any DROP/TRUNCATE token also matched
+        // "ALTER TABLE t DROP COLUMN c" and "CREATE TABLE ... DROP ..." in identifiers
+        // or comments, which would make DISABLE_DROP_TRUNCATE=true block legitimate,
+        // non-destructive schema evolution.
+        for (Token token : tokensList) {
+            if (token.getChannel() != Token.DEFAULT_CHANNEL) {
+                // Skip whitespace and comments.
+                continue;
+            }
+            if (token.getType() == Token.EOF) {
+                break;
+            }
+            return token.getType() == MySqlParser.DROP
+                    || token.getType() == MySqlParser.TRUNCATE;
         }
 
-        return result;
+        return false;
     }
 
     /**