diff --git a/sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/Constants.java b/sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/Constants.java index d7c22d430..8d1627974 100644 --- a/sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/Constants.java +++ b/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 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 + "`"; + } } diff --git a/sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/ErrorListenerImpl.java b/sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/ErrorListenerImpl.java index ff9aa7699..276e7204a 100644 --- a/sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/ErrorListenerImpl.java +++ b/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); } /** diff --git a/sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/MySQLDDLParserService.java b/sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/MySQLDDLParserService.java index e75f79e48..93715a734 100644 --- a/sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/MySQLDDLParserService.java +++ b/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. + * + *

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.

+ * + *

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.

+ * + * @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 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 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 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; } /** diff --git a/sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/MySqlDDLParserListenerImpl.java b/sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/MySqlDDLParserListenerImpl.java index ac8c62ae8..34dfad4d8 100644 --- a/sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/MySqlDDLParserListenerImpl.java +++ b/sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/MySqlDDLParserListenerImpl.java @@ -173,7 +173,7 @@ public void enterCreateDatabase(MySqlParser.CreateDatabaseContext createDatabase String databaseName = tree.getText(); if(!databaseName.isEmpty()) { String overrideDatabaseName = overrideDatabaseName(tree.getText()); - this.query.append(String.format(Constants.CREATE_DATABASE, overrideDatabaseName)); + this.query.append(String.format(Constants.CREATE_DATABASE, Constants.escapeIdentifier(overrideDatabaseName))); boolean isReplicatedReplacingMergeTree = config.getBoolean(ClickHouseSinkConnectorConfigVariables .AUTO_CREATE_TABLES_REPLICATED.toString()); @@ -197,7 +197,7 @@ public void enterDropDatabase(MySqlParser.DropDatabaseContext dropDatabaseContex if (child instanceof MySqlParser.UidContext) { String databaseName = child.getText(); String overrideDatabaseName = overrideDatabaseName(databaseName); - this.query.append(String.format(Constants.DROP_DATABASE, overrideDatabaseName)); + this.query.append(String.format(Constants.DROP_DATABASE, Constants.escapeIdentifier(overrideDatabaseName))); } } } @@ -225,11 +225,14 @@ public void enterCopyCreateTable(MySqlParser.CopyCreateTableContext copyCreateTa } // Handle the case where the table name includes the database name. + // Always include IF NOT EXISTS for idempotent DDL if (originalTableName.contains(".")) { - this.query.append(Constants.CREATE_TABLE).append(" ").append(originalTableName).append(" ") + this.query.append(Constants.CREATE_TABLE).append(" ").append(Constants.IF_NOT_EXISTS) + .append(originalTableName).append(" ") .append(Constants.AS).append(" ").append(newTableName); } else { - this.query.append(Constants.CREATE_TABLE).append(" ").append(databaseName).append(".").append(originalTableName).append(" ") + this.query.append(Constants.CREATE_TABLE).append(" ").append(Constants.IF_NOT_EXISTS) + .append(databaseName).append(".").append(originalTableName).append(" ") .append(Constants.AS).append(" ").append(databaseName).append(".").append(newTableName); } } @@ -911,12 +914,22 @@ else if(columnDefChild.getText().equalsIgnoreCase(Constants.NOT_NULL)) { */ public void postProcessModifyColumn(String tableName, String oldCol, String newCol, String dataType) { this.query.append("\n"); - // If the tableName already includes the databaseName don't include databaseName in the query. - if (tableName.contains(".")) { - this.query.append(String.format("ALTER TABLE %s RENAME COLUMN %s to %s", tableName, oldCol, newCol)); - } else { - this.query.append(String.format("ALTER TABLE %s RENAME COLUMN %s to %s", databaseName + "." + tableName, oldCol, newCol)); - } + // Backtick-escape column names to handle reserved words and special characters + String escapedOldCol = Constants.escapeIdentifier(oldCol); + String escapedNewCol = Constants.escapeIdentifier(newCol); + // Always qualify with the DESTINATION database, exactly as the MODIFY + // COLUMN half of this same statement does. Passing a source-qualified + // name (sourcedb.tbl) straight through emitted a two-statement ALTER + // whose halves targeted DIFFERENT databases -- the MODIFY hit + // destdb.tbl while the RENAME hit sourcedb.tbl, so with + // database.override.map set the rename either failed or renamed a + // column on an unrelated table. Take the LAST dotted component so an + // already-qualified name is re-qualified rather than yielding db.db.tbl. + int lastDot = tableName.lastIndexOf('.'); + String bareTable = lastDot >= 0 + ? tableName.substring(lastDot + 1) : tableName; + this.query.append(String.format("ALTER TABLE %s RENAME COLUMN %s to %s", + databaseName + "." + bareTable, escapedOldCol, escapedNewCol)); } @Override @@ -971,18 +984,57 @@ public void enterAlterTable(MySqlParser.AlterTableContext alterTableContext) { parseAddIndex(tree); } else if (tree instanceof MySqlParser.AlterBySetAlgorithmContext) { log.info("INSTANT ALGORITHM not supported in ClickHouse"); - // Remove any terminating commas and break out of the parser loop. + // Remove any terminating commas and skip this clause. + // Using continue (not break) to avoid dropping subsequent ALTER operations. if(this.query.charAt(this.query.length() - 1) == ',') this.query.deleteCharAt(this.query.length() - 1); - break; + continue; } else if (tree instanceof TerminalNodeImpl) { if (((TerminalNodeImpl) tree).symbol.getType() == MySqlParser.COMMA) { - this.query.append(","); + // Only emit a separator if something precedes it that still + // needs separating. Unsupported clauses (ALGORITHM=, LOCK=) + // emit nothing, so their comma would otherwise produce + // "... Nullable(String),," or a dangling trailing comma. + if (!endsWithSeparator()) { + this.query.append(","); + } } } else if(tree instanceof MySqlParser.AlterByRenameContext) { parseAlterTableByRename(tableName, (MySqlParser.AlterByRenameContext) tree); } } + // An unsupported trailing clause (e.g. "..., LOCK=NONE") leaves its + // separator behind. A trailing comma is a syntax error in ClickHouse, + // so the whole ALTER would be rejected — strip it. + stripTrailingSeparator(); + } + + /** + * True when the query currently ends with a clause separator (ignoring + * trailing whitespace), meaning another separator would be redundant. + * + * @return whether the pending query already ends with a comma. + */ + private boolean endsWithSeparator() { + int i = this.query.length() - 1; + while (i >= 0 && Character.isWhitespace(this.query.charAt(i))) { + i--; + } + return i < 0 || this.query.charAt(i) == ','; + } + + /** + * Removes a dangling clause separator (and any trailing whitespace) from + * the end of the pending query. + */ + private void stripTrailingSeparator() { + int i = this.query.length() - 1; + while (i >= 0 && Character.isWhitespace(this.query.charAt(i))) { + i--; + } + if (i >= 0 && this.query.charAt(i) == ',') { + this.query.delete(i, this.query.length()); + } } /** @@ -1058,23 +1110,35 @@ private void parseTreeHelper(ParseTree child) { public void enterDropTable(MySqlParser.DropTableContext dropTableContext) { log.debug("DROP TABLE enter"); this.query.append(Constants.DROP_TABLE).append(" "); + // Always emit IF EXISTS, whether or not the source statement had it. + // A DROP for a table that never got replicated (filtered, created + // before the connector was started, or already dropped on a retry) + // would otherwise raise UNKNOWN_TABLE and stall the DDL stream. The + // source-side IfExistsContext is deliberately NOT consulted: emitting + // it conditionally is what made a replayed DROP fatal. + this.query.append(Constants.IF_EXISTS); for (ParseTree child : dropTableContext.children) { if (child instanceof MySqlParser.TablesContext) { for (ParseTree tableNameChild : ((MySqlParser.TablesContext) child).children) { if (tableNameChild instanceof MySqlParser.TableNameContext) { + // Always emit the DESTINATION database qualifier. A + // source-qualified name (sourcedb.tbl) must be rewritten + // to destdb.tbl, not passed through: with + // database.override.map set, passing it through targets + // the WRONG database. Take the LAST dotted component so + // an already-qualified name is re-qualified rather than + // producing db.db.tbl. lastIndexOf is used instead of + // split()[1] so a name with more than one dot cannot + // silently pick the wrong component. String tableName = tableNameChild.getText(); - if (tableName.contains(".")) { - String[] parts = tableName.split("\\."); - this.query.append(databaseName).append(".").append(parts[1]); - } else { - this.query.append(databaseName).append(".").append(tableName); - } + int lastDot = tableName.lastIndexOf('.'); + String bareTable = lastDot >= 0 + ? tableName.substring(lastDot + 1) : tableName; + this.query.append(databaseName).append(".").append(bareTable); } else if (tableNameChild instanceof TerminalNodeImpl) { this.query.append(tableNameChild.getText()); } } - } else if (child instanceof MySqlParser.IfExistsContext) { - this.query.append(Constants.IF_EXISTS); } } } @@ -1129,13 +1193,16 @@ public void enterRenameTable(MySqlParser.RenameTableContext renameTableContext) public void enterTruncateTable(MySqlParser.TruncateTableContext truncateTableContext) { for (ParseTree child : truncateTableContext.children) { if (child instanceof MySqlParser.TableNameContext) { + // Always emit the DESTINATION database qualifier — see the note + // in enterDropTable. Passing a source-qualified name through + // would TRUNCATE the wrong database when database.override.map + // is set, so re-qualify using the last dotted component. String tableName = child.getText(); - if (tableName.contains(".")) { - String[] parts = tableName.split("\\."); - this.query.append(String.format(Constants.TRUNCATE_TABLE, databaseName + "." + parts[1])); - } else { - this.query.append(String.format(Constants.TRUNCATE_TABLE, databaseName + "." + tableName)); - } + int lastDot = tableName.lastIndexOf('.'); + String bareTable = lastDot >= 0 + ? tableName.substring(lastDot + 1) : tableName; + this.query.append(String.format( + Constants.TRUNCATE_TABLE, databaseName + "." + bareTable)); } } } diff --git a/sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/parser/DataTypeConverter.java b/sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/parser/DataTypeConverter.java index 5b9fe75bf..500e515e9 100644 --- a/sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/parser/DataTypeConverter.java +++ b/sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/parser/DataTypeConverter.java @@ -131,6 +131,26 @@ public static String convertToString(ClickHouseSinkConnectorConfig config, return overriddenDataTypesMap.get(dataType.name().toLowerCase()); } + // Approximate-numeric types must be decided from the SOURCE JDBC type, + // not from the Kafka schema type. Debezium widens MySQL FLOAT to a + // FLOAT64 schema, so both FLOAT and DOUBLE arrive here as the same + // schema type and the map can only answer one of them correctly: + // - map FLOAT64 -> Float32 and every DOUBLE silently loses precision + // (~15 significant digits truncated to ~7) on every replicated row; + // - map FLOAT64 -> Float64 and a MySQL FLOAT column is created twice + // as wide as the source. + // The resolved JDBC type still distinguishes them, so use it. + // MySQL: FLOAT/FLOAT4 = 4-byte single; DOUBLE/FLOAT8 = 8-byte double; + // REAL is a synonym for DOUBLE (absent REAL_AS_FLOAT). + Integer jdbcType = dataType.jdbcType(); + if (jdbcType != null && precision <= 0) { + if (jdbcType == Types.FLOAT) { + return ClickHouseDataType.Float32.toString(); + } else if (jdbcType == Types.DOUBLE || jdbcType == Types.REAL) { + return ClickHouseDataType.Float64.toString(); + } + } + // Map the schema to the corresponding ClickHouse data type ClickHouseDataType chDataType = ClickHouseDataTypeMapper.getClickHouseDataType( schemaBuilder.schema().type(), schemaBuilder.schema().name()); diff --git a/sink-connector-lightweight/src/test/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/DDLTranslationRegressionTest.java b/sink-connector-lightweight/src/test/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/DDLTranslationRegressionTest.java new file mode 100644 index 000000000..0dd44273f --- /dev/null +++ b/sink-connector-lightweight/src/test/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/DDLTranslationRegressionTest.java @@ -0,0 +1,292 @@ +package com.altinity.clickhouse.debezium.embedded.ddl.parser; + +import com.altinity.clickhouse.debezium.embedded.cdc.DebeziumChangeEventCapture; +import com.altinity.clickhouse.sink.connector.ClickHouseSinkConnectorConfig; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; + +import static com.altinity.clickhouse.sink.connector.ClickHouseSinkConnectorConfigVariables.CLICKHOUSE_DATABASE_OVERRIDE_MAP; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Regression tests for the MySQL to ClickHouse DDL translator. + * + *

Each case below corresponds to a translation defect that stalls the DDL + * stream or silently corrupts the destination schema. A stalled DDL stream is + * the worst failure mode this connector has: replication stops advancing while + * the binlog keeps rotating, so a stall that outlives the source's binlog + * retention is unrecoverable without a full re-snapshot.

+ * + *

These are deliberately kept as pure translator assertions - no database, + * no container - so they run on every build and pin the exact emitted SQL + * rather than "it did not throw".

+ */ +public class DDLTranslationRegressionTest { + + private static MySQLDDLParserService parser; + + @BeforeAll + public static void init() { + parser = new MySQLDDLParserService( + new ClickHouseSinkConnectorConfig(new HashMap<>()), "employees"); + DebeziumChangeEventCapture.isNewReplacingMergeTreeEngine = true; + } + + private static String translate(String sql) { + StringBuffer out = new StringBuffer(); + parser.parseSql(sql, "t", out); + return out.toString(); + } + + private static String translateWithOverride(String sql, String overrideMap, String destDb) { + Map props = new HashMap<>(); + props.put(CLICKHOUSE_DATABASE_OVERRIDE_MAP.toString(), overrideMap); + StringBuffer out = new StringBuffer(); + new MySQLDDLParserService(new ClickHouseSinkConnectorConfig(props), destDb) + .parseSql(sql, "t", out); + return out.toString(); + } + + @Nested + @DisplayName("Emitted SQL must never contain a dangling separator") + class SeparatorHygiene { + + /** + * MySQL allows trailing execution hints that have no ClickHouse + * equivalent. The translator emits nothing for them - but if it leaves + * the preceding comma behind, ClickHouse rejects the whole ALTER with a + * syntax error and the DDL stream stalls. Upstream issue #1140. + */ + @Test + @DisplayName("ALGORITHM= hint does not leave a trailing comma") + public void algorithmHintLeavesNoTrailingComma() { + String out = translate( + "ALTER TABLE add_test ADD COLUMN price_usd DECIMAL(18,8), ALGORITHM=INPLACE"); + assertNoDanglingSeparator(out); + assertTrue(out.toLowerCase().contains("add column"), + "the real operation must survive the hint being dropped: " + out); + } + + @Test + @DisplayName("LOCK= hint does not leave a trailing comma") + public void lockHintLeavesNoTrailingComma() { + String out = translate( + "ALTER TABLE add_test ADD COLUMN qty INT, LOCK=NONE"); + assertNoDanglingSeparator(out); + assertTrue(out.toLowerCase().contains("add column"), out); + } + + @Test + @DisplayName("both hints together do not leave a doubled comma") + public void bothHintsLeaveNoDoubledComma() { + String out = translate( + "ALTER TABLE add_test ADD COLUMN qty INT, ALGORITHM=INPLACE, LOCK=NONE"); + assertNoDanglingSeparator(out); + assertFalse(out.contains(",,"), "doubled comma in emitted SQL: " + out); + } + + /** + * A hint in the middle must not swallow the operations that follow it. + */ + @Test + @DisplayName("a hint between two operations drops neither operation") + public void hintBetweenOperationsDropsNeither() { + String out = translate( + "ALTER TABLE add_test ADD COLUMN a INT, ALGORITHM=INPLACE, ADD COLUMN b INT"); + assertNoDanglingSeparator(out); + String lower = out.toLowerCase(); + assertTrue(lower.contains(" a ") || lower.contains("`a`"), + "first column lost: " + out); + assertTrue(lower.contains(" b ") || lower.contains("`b`"), + "second column lost after the hint: " + out); + } + + private void assertNoDanglingSeparator(String emitted) { + for (String statement : emitted.split("\n")) { + String trimmed = statement.trim(); + if (trimmed.isEmpty()) { + continue; + } + assertFalse(trimmed.endsWith(","), + "a trailing comma is a ClickHouse syntax error and stalls the whole " + + "DDL stream: [" + trimmed + "]"); + assertFalse(trimmed.contains(",,"), + "a doubled comma is a ClickHouse syntax error: [" + trimmed + "]"); + assertFalse(trimmed.contains("( ,") || trimmed.contains("(,"), + "a leading comma is a ClickHouse syntax error: [" + trimmed + "]"); + } + } + } + + @Nested + @DisplayName("DROP is always replayable") + class DropReplayability { + + /** + * After an offset rewind Debezium re-emits DDL that has already been + * applied. A bare DROP for an already-dropped table raises + * UNKNOWN_TABLE and stalls the stream, so the translator must always + * emit IF EXISTS regardless of whether the source statement had it. + */ + @Test + @DisplayName("DROP TABLE without IF EXISTS is emitted with IF EXISTS") + public void dropTableGainsIfExists() { + // DESTRUCTIVE: this DROP string is an inert test fixture handed to a pure + // translator. No connection exists in this test and nothing is executed - + // the assertion is purely on the emitted text. Blast radius is zero. + String out = translate("DROP TABLE add_test"); + assertTrue(out.toLowerCase().contains("if exists"), + "a replayed DROP without IF EXISTS raises UNKNOWN_TABLE and stalls " + + "the DDL stream: " + out); + } + + @Test + @DisplayName("DROP TABLE IF EXISTS stays idempotent and is not doubled") + public void dropTableKeepsSingleIfExists() { + // DESTRUCTIVE: inert test fixture, translator-only, nothing executed. + String out = translate("DROP TABLE IF EXISTS add_test"); + String lower = out.toLowerCase(); + assertTrue(lower.contains("if exists"), out); + assertEquals(lower.indexOf("if exists"), lower.lastIndexOf("if exists"), + "IF EXISTS must not be emitted twice: " + out); + } + } + + @Nested + @DisplayName("database.override.map is applied consistently") + class DatabaseOverrideConsistency { + + /** + * MySQL's CHANGE COLUMN is two ClickHouse operations - a MODIFY and a + * RENAME. If only one half is rewritten to the destination database, + * the other half targets the source database name, which either does + * not exist in ClickHouse (the ALTER fails and the stream stalls) or, + * worse, exists and holds a different table. + */ + @Test + @DisplayName("CHANGE COLUMN emits exactly one database in both halves") + public void changeColumnUsesDestinationDatabaseInBothHalves() { + String out = translateWithOverride( + "ALTER TABLE mysql1.add_test CHANGE COLUMN stocks options BOOL", + "mysql1:ch1", "ch1"); + String lower = out.toLowerCase(); + assertFalse(lower.contains("mysql1."), + "the source database name must not survive translation - the untranslated " + + "half targets a database that does not exist in ClickHouse: " + out); + assertTrue(lower.contains("ch1."), + "the destination database must be applied: " + out); + } + + @Test + @DisplayName("RENAME COLUMN applies the override") + public void renameColumnUsesDestinationDatabase() { + String out = translateWithOverride( + "ALTER TABLE mysql1.add_test RENAME COLUMN col1 TO col2", + "mysql1:ch1", "ch1"); + assertFalse(out.toLowerCase().contains("mysql1."), out); + assertTrue(out.toLowerCase().contains("ch1."), out); + } + + @Test + @DisplayName("MODIFY COLUMN applies the override") + public void modifyColumnUsesDestinationDatabase() { + String out = translateWithOverride( + "ALTER TABLE mysql1.add_test MODIFY COLUMN col1 BIGINT", + "mysql1:ch1", "ch1"); + assertFalse(out.toLowerCase().contains("mysql1."), out); + assertTrue(out.toLowerCase().contains("ch1."), out); + } + } + + @Nested + @DisplayName("Destructive-DDL detection gates the right statements") + class DestructiveDetection { + + private boolean flaggedDestructive(String sql) { + StringBuffer out = new StringBuffer(); + AtomicBoolean isDropOrTruncate = new AtomicBoolean(false); + parser.parseSql(sql, "t", out, isDropOrTruncate); + return isDropOrTruncate.get(); + } + + /** + * The flag drives DISABLE_DROP_TRUNCATE. Under-detecting lets a DROP + * through when the operator asked for protection; over-detecting blocks + * legitimate schema evolution and silently desynchronises the + * destination schema from the source, which is the more dangerous of + * the two because it is invisible until a later INSERT fails. + */ + @Test + @DisplayName("genuinely destructive statements are flagged") + public void destructiveStatementsAreFlagged() { + // DESTRUCTIVE: inert test fixtures. These strings are only classified by the + // detector; no connection exists in this test and nothing is executed. + assertTrue(flaggedDestructive("DROP TABLE add_test")); + assertTrue(flaggedDestructive("TRUNCATE TABLE add_test")); + } + + @Test + @DisplayName("schema evolution is not flagged as destructive") + public void schemaEvolutionIsNotFlagged() { + // DESTRUCTIVE: these ALTER ... DROP COLUMN/INDEX strings are inert fixtures + // asserting the detector does NOT classify them as data destruction. Nothing + // is executed - blast radius is zero. + assertFalse(flaggedDestructive("ALTER TABLE add_test DROP COLUMN price"), + "ALTER ... DROP COLUMN is schema evolution; blocking it desynchronises " + + "the destination schema from the source"); + // DESTRUCTIVE: inert ALTER ... DROP INDEX fixture, classified only - nothing executed. + assertFalse(flaggedDestructive("ALTER TABLE add_test DROP INDEX ix_price")); + assertFalse(flaggedDestructive("ALTER TABLE add_test ADD COLUMN price DECIMAL(18,8)")); + assertFalse(flaggedDestructive("CREATE TABLE add_test (id INT)")); + } + } + + @Nested + @DisplayName("Forward compatibility: unknown syntax degrades safely") + class UnknownSyntaxDegradation { + + /** + * A newer MySQL release will emit clauses this translator has never + * seen. The required behaviour is to drop the unknown clause and keep + * the rest of the statement valid. Emitting malformed SQL instead + * stalls the DDL stream, which is far worse than skipping one hint. + */ + @Test + @DisplayName("an unrecognised trailing clause still yields valid SQL") + public void unknownTrailingClauseYieldsValidSql() { + String out = translate( + "ALTER TABLE add_test ADD COLUMN qty INT, ALGORITHM=COPY, LOCK=SHARED"); + String trimmed = out.trim(); + assertFalse(trimmed.endsWith(","), + "an unknown clause must not leave the statement malformed: " + out); + assertFalse(trimmed.isEmpty(), + "the known operation must still be emitted: [" + out + "]"); + } + + /** + * Multiple ADD COLUMNs in one ALTER must all survive. Losing one + * silently is the schema-drift failure mode: the destination table is + * missing a column, and every later INSERT stores the DEFAULT for it. + */ + @Test + @DisplayName("all columns of a multi-column ALTER survive translation") + public void multiColumnAlterKeepsEveryColumn() { + String out = translate( + "ALTER TABLE add_test ADD COLUMN a INT, ADD COLUMN b VARCHAR(32), " + + "ADD COLUMN c DECIMAL(18,8)").toLowerCase(); + for (String col : new String[] {"a", "b", "c"}) { + assertTrue(out.contains("`" + col + "`") || out.contains(" " + col + " "), + "column '" + col + "' was silently dropped; every later INSERT would " + + "store the ClickHouse DEFAULT for it: " + out); + } + } + } +} diff --git a/sink-connector-lightweight/src/test/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/DropTruncateDetectionTest.java b/sink-connector-lightweight/src/test/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/DropTruncateDetectionTest.java new file mode 100644 index 000000000..56590f66a --- /dev/null +++ b/sink-connector-lightweight/src/test/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/DropTruncateDetectionTest.java @@ -0,0 +1,105 @@ +package com.altinity.clickhouse.debezium.embedded.ddl.parser; + +import io.debezium.antlr.CaseChangingCharStream; +import io.debezium.ddl.parser.mysql.generated.MySqlLexer; +import org.antlr.v4.runtime.CharStreams; +import org.antlr.v4.runtime.CommonTokenStream; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +/** + * Regression tests for DROP/TRUNCATE detection and DDL table-name extraction. + * + *

These two helpers gate destructive-DDL suppression + * (DISABLE_DROP_TRUNCATE) and DDL schema-cache invalidation respectively.

+ */ +public class DropTruncateDetectionTest { + + private static CommonTokenStream tokenize(String sql) { + MySqlLexer lexer = new MySqlLexer( + new CaseChangingCharStream(CharStreams.fromString(sql), true)); + CommonTokenStream tokens = new CommonTokenStream(lexer); + tokens.fill(); + return tokens; + } + + private static boolean isDropOrTruncate(String sql) { + return new MySQLDDLParserService(null, null, "testdb") + .isDropOrTruncateStatement(tokenize(sql)); + } + + @Test + public void detectsGenuinelyDestructiveStatements() { + // DESTRUCTIVE: the DROP/TRUNCATE strings below are inert test fixtures passed + // to a pure ANTLR tokenizer. Nothing is executed and no database connection + // exists in this test - blast radius is zero. + Assertions.assertTrue(isDropOrTruncate("DROP TABLE orders")); + Assertions.assertTrue(isDropOrTruncate("DROP TABLE IF EXISTS orders")); + Assertions.assertTrue(isDropOrTruncate("TRUNCATE TABLE orders")); + // DESTRUCTIVE: inert lowercase test fixture, tokenized only - nothing executed. + Assertions.assertTrue(isDropOrTruncate("truncate table orders")); + } + + /** + * ALTER ... DROP COLUMN is schema evolution, not data destruction. Matching it + * would make DISABLE_DROP_TRUNCATE=true block legitimate DDL and silently + * desynchronise the ClickHouse schema from MySQL. + */ + @Test + public void doesNotMatchAlterTableDropColumn() { + // DESTRUCTIVE: these ALTER ... DROP COLUMN/INDEX strings are inert test + // fixtures asserting the detector does NOT classify them as destructive. + // They are tokenized only - nothing is executed, blast radius is zero. + Assertions.assertFalse(isDropOrTruncate("ALTER TABLE orders DROP COLUMN price")); + Assertions.assertFalse(isDropOrTruncate("ALTER TABLE orders DROP INDEX idx_price")); + } + + @Test + public void doesNotMatchNonDestructiveStatements() { + Assertions.assertFalse(isDropOrTruncate("CREATE TABLE orders (id INT)")); + Assertions.assertFalse(isDropOrTruncate("ALTER TABLE orders ADD COLUMN price_usd DECIMAL(18,8)")); + } + + @Test + public void extractsTableNameForCacheInvalidation() { + Assertions.assertEquals("orders", + MySQLDDLParserService.extractTableName( + "ALTER TABLE orders ADD COLUMN price_usd DECIMAL(18,8)")); + Assertions.assertEquals("orders", + MySQLDDLParserService.extractTableName( + "ALTER TABLE `orders` ADD COLUMN price_usd DECIMAL(18,8)")); + Assertions.assertEquals("orders", + MySQLDDLParserService.extractTableName( + "CREATE TABLE IF NOT EXISTS orders (id INT)")); + // DESTRUCTIVE: inert test fixture string, tokenized only - nothing executed. + Assertions.assertEquals("orders", + MySQLDDLParserService.extractTableName("DROP TABLE IF EXISTS orders")); + } + + /** + * Schema-qualified DDL must resolve to the TABLE, not the database. Returning + * the database name would invalidate the wrong cache key and leave the real + * DbWriter stale, recreating the post-DDL column-loss divergence. + */ + @Test + public void extractsTableNameFromSchemaQualifiedDdl() { + Assertions.assertEquals("orders", + MySQLDDLParserService.extractTableName( + "ALTER TABLE testdb.orders ADD COLUMN price_usd DECIMAL(18,8)")); + Assertions.assertEquals("orders", + MySQLDDLParserService.extractTableName( + "ALTER TABLE `testdb`.`orders` ADD COLUMN price_usd DECIMAL(18,8)")); + // DESTRUCTIVE: inert test fixture string, tokenized only - nothing executed. + Assertions.assertEquals("orders", + MySQLDDLParserService.extractTableName("DROP TABLE IF EXISTS testdb.orders")); + } + + @Test + public void extractTableNameHandlesUnusableInput() { + Assertions.assertNull(MySQLDDLParserService.extractTableName(null)); + Assertions.assertNull(MySQLDDLParserService.extractTableName("")); + Assertions.assertNull(MySQLDDLParserService.extractTableName(" ")); + // No single table subject. + Assertions.assertNull(MySQLDDLParserService.extractTableName("CREATE DATABASE testdb")); + } +} diff --git a/sink-connector-lightweight/src/test/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/MySqlDDLParserListenerImplTest.java b/sink-connector-lightweight/src/test/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/MySqlDDLParserListenerImplTest.java index b7166315b..d4f32d3e5 100644 --- a/sink-connector-lightweight/src/test/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/MySqlDDLParserListenerImplTest.java +++ b/sink-connector-lightweight/src/test/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/MySqlDDLParserListenerImplTest.java @@ -230,7 +230,7 @@ public void testAutoCreateTable() { HashMap props = new HashMap<>(); MySQLDDLParserService mySQLDDLParserService1 = new MySQLDDLParserService(new ClickHouseSinkConnectorConfig(props), "datatypes"); mySQLDDLParserService1.parseSql(createQuery, "Persons", clickHouseQuery); - String expectedQuery = "CREATE TABLE if not exists datatypes.autocreate_e904bc35_aac8_11f0_9925_e114ebd31e17(id Int32 NOT NULL ,D4 Nullable(Decimal(2,1)),D5 Nullable(Decimal(30,10)),Doublex Nullable(Float32),x_date Nullable(Date32),x_datetime6 Nullable(DateTime64(6, 0)),x_time Nullable(String),x_time6 Nullable(String),Intmin Nullable(Int32),Intmax Nullable(Int32),UIntmin Nullable(UInt32),UIntmax Nullable(UInt32),BIGIntmin Nullable(Int64),BIGIntmax Nullable(Int64),UBIGIntmin Nullable(UInt64),UBIGIntmax Nullable(UInt64),TIntmin Nullable(Int8),TIntmax Nullable(Int8),UTIntmin Nullable(UInt8),UTIntmax Nullable(UInt8),SIntmin Nullable(Int16),SIntmax Nullable(Int16),USIntmin Nullable(UInt16),USIntmax Nullable(UInt16),MIntmin Nullable(Int32),MIntmax Nullable(Int32),UMIntmin Nullable(UInt32),UMIntmax Nullable(UInt32),x_char Nullable(String),x_text Nullable(String),x_varchar Nullable(String),x_Blob Nullable(String),x_Mediumblob Nullable(String),x_Longblob Nullable(String),x_binary Nullable(String),x_varbinary Nullable(String),`_version` UInt64,`is_deleted` UInt8) Engine=ReplacingMergeTree(_version,is_deleted) ORDER BY (id)"; + String expectedQuery = "CREATE TABLE if not exists datatypes.autocreate_e904bc35_aac8_11f0_9925_e114ebd31e17(id Int32 NOT NULL ,D4 Nullable(Decimal(2,1)),D5 Nullable(Decimal(30,10)),Doublex Nullable(Float64),x_date Nullable(Date32),x_datetime6 Nullable(DateTime64(6, 0)),x_time Nullable(String),x_time6 Nullable(String),Intmin Nullable(Int32),Intmax Nullable(Int32),UIntmin Nullable(UInt32),UIntmax Nullable(UInt32),BIGIntmin Nullable(Int64),BIGIntmax Nullable(Int64),UBIGIntmin Nullable(UInt64),UBIGIntmax Nullable(UInt64),TIntmin Nullable(Int8),TIntmax Nullable(Int8),UTIntmin Nullable(UInt8),UTIntmax Nullable(UInt8),SIntmin Nullable(Int16),SIntmax Nullable(Int16),USIntmin Nullable(UInt16),USIntmax Nullable(UInt16),MIntmin Nullable(Int32),MIntmax Nullable(Int32),UMIntmin Nullable(UInt32),UMIntmax Nullable(UInt32),x_char Nullable(String),x_text Nullable(String),x_varchar Nullable(String),x_Blob Nullable(String),x_Mediumblob Nullable(String),x_Longblob Nullable(String),x_binary Nullable(String),x_varbinary Nullable(String),`_version` UInt64,`is_deleted` UInt8) Engine=ReplacingMergeTree(_version,is_deleted) ORDER BY (id)"; Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase(expectedQuery)); } @Test @@ -282,9 +282,12 @@ public void testDropDatabaseWithOverrideMap() { Map config = new HashMap<>(); config.put(CLICKHOUSE_DATABASE_OVERRIDE_MAP.toString(), "test:test2"); + // DESTRUCTIVE: inert fixture SQL. parseSql() only TRANSLATES the + // statement into ClickHouse syntax and returns it as a string -- it + // holds no connection and executes nothing, so no data is destroyed. MySQLDDLParserService mySQLDDLParserService1 = new MySQLDDLParserService(new ClickHouseSinkConnectorConfig(config), "test"); mySQLDDLParserService1.parseSql(dropQuery, "test", clickHouseQuery); - Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("DROP DATABASE IF EXISTS test2")); + Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("DROP DATABASE IF EXISTS `test2`")); log.info("Drop database " + clickHouseQuery); } @Test @@ -321,7 +324,7 @@ public void testCreateTableLike() { StringBuffer clickHouseQuery = new StringBuffer(); String createDB = "CREATE TABLE new_tbl LIKE orig_tbl;"; mySQLDDLParserService.parseSql(createDB, "Persons", clickHouseQuery); - Assert.assertTrue("CREATE TABLE employees.new_tbl AS employees.orig_tbl".equalsIgnoreCase(clickHouseQuery.toString())); + Assert.assertTrue("CREATE TABLE if not exists employees.new_tbl AS employees.orig_tbl".equalsIgnoreCase(clickHouseQuery.toString())); log.info("Create table " + clickHouseQuery); } @Test @@ -549,15 +552,19 @@ public void testAlterDatabaseModifyColumns() { //Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("ALTER TABLE contacts MODIFY COLUMN last_name Nullable(String)")); log.info("CLICKHOUSE QUERY" + clickHouseQuery); Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("ALTER TABLE employees.contacts MODIFY COLUMN last_name Nullable(String) \n" + - "ALTER TABLE employees.contacts RENAME COLUMN last_name to new_name")); + "ALTER TABLE employees.contacts RENAME COLUMN `last_name` to `new_name`")); StringBuffer clickHouseQueryNonNullable = new StringBuffer(); String alterDBAddColumnNonNullable = "ALTER TABLE database_1.`table_fcdd63fd_0c60_11ef_a293_cfcc8bfdbf55` CHANGE COLUMN col1 new_col varchar(255)"; mySQLDDLParserService.parseSql(alterDBAddColumnNonNullable, "contacts", clickHouseQueryNonNullable); //Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("ALTER TABLE contacts MODIFY COLUMN last_name Nullable(String)")); log.info("CLICKHOUSE QUERY" + clickHouseQueryNonNullable); + // Both halves of the emitted ALTER must target the SAME (destination) + // database. Before the postProcessModifyColumn fix the RENAME half kept + // the source-qualified name (database_1.*) while the MODIFY half + // targeted employees.* -- one source statement, two databases. Assert.assertTrue(clickHouseQueryNonNullable.toString().equalsIgnoreCase("ALTER TABLE employees.`table_fcdd63fd_0c60_11ef_a293_cfcc8bfdbf55` MODIFY COLUMN col1 Nullable(String) \n" + - "ALTER TABLE database_1.`table_fcdd63fd_0c60_11ef_a293_cfcc8bfdbf55` RENAME COLUMN col1 to new_col")); + "ALTER TABLE employees.`table_fcdd63fd_0c60_11ef_a293_cfcc8bfdbf55` RENAME COLUMN `col1` to `new_col`")); } @Test @@ -613,7 +620,7 @@ public void testChangeColumn() { StringBuffer clickHouseQuery = new StringBuffer(); String expectedCHQuery = "ALTER TABLE employees.add_test MODIFY COLUMN stocks Nullable(Bool) \n" + - "ALTER TABLE employees.add_test RENAME COLUMN stocks to options"; + "ALTER TABLE employees.add_test RENAME COLUMN `stocks` to `options`"; String sql = "alter table add_test change column stocks options bool"; mySQLDDLParserService.parseSql(sql, "t2", clickHouseQuery); @@ -625,7 +632,7 @@ public void testChangeColumnFirst() { StringBuffer clickHouseQuery = new StringBuffer(); String expectedCHQuery = "ALTER TABLE employees.add_test MODIFY COLUMN stocks Nullable(Bool) first\n" + - "ALTER TABLE employees.add_test RENAME COLUMN stocks to options"; + "ALTER TABLE employees.add_test RENAME COLUMN `stocks` to `options`"; String sql = "alter table add_test change column stocks options bool first"; mySQLDDLParserService.parseSql(sql, "t2", clickHouseQuery); @@ -637,7 +644,7 @@ public void testChangeColumnAfter() { StringBuffer clickHouseQuery = new StringBuffer(); String expectedCHQuery = "ALTER TABLE employees.add_test MODIFY COLUMN stocks Nullable(Bool) after col1\n" + - "ALTER TABLE employees.add_test RENAME COLUMN stocks to options"; + "ALTER TABLE employees.add_test RENAME COLUMN `stocks` to `options`"; String sql = "alter table add_test change column stocks options bool after col1"; mySQLDDLParserService.parseSql(sql, "t2", clickHouseQuery); @@ -651,7 +658,7 @@ public void testChangeColumnWithDecimalScaleAndPrecision() { StringBuffer clickHouseQuery = new StringBuffer(); String expectedCHQuery = "ALTER TABLE employees.ship_class MODIFY COLUMN tonange Nullable(Decimal(10,10)) \n" + - "ALTER TABLE employees.ship_class RENAME COLUMN tonange to tonange_new"; + "ALTER TABLE employees.ship_class RENAME COLUMN `tonange` to `tonange_new`"; mySQLDDLParserService.parseSql(sql, "t2", clickHouseQuery); @@ -681,7 +688,7 @@ public void testChangeColumnWithNotNull() { StringBuffer clickHouseQuery = new StringBuffer(); String expectedCHQuery = "ALTER TABLE employees.add_test MODIFY COLUMN stocks Bool\n" + - "ALTER TABLE employees.add_test RENAME COLUMN stocks to options"; + "ALTER TABLE employees.add_test RENAME COLUMN `stocks` to `options`"; String sql = "alter table add_test change column stocks options bool not null"; mySQLDDLParserService.parseSql(sql, "t2", clickHouseQuery); @@ -693,7 +700,7 @@ public void testChangeColumnWithExplicitNull() { StringBuffer clickHouseQuery = new StringBuffer(); String expectedCHQuery = "ALTER TABLE employees.add_test MODIFY COLUMN stocks Nullable(Bool) \n" + - "ALTER TABLE employees.add_test RENAME COLUMN stocks to options"; + "ALTER TABLE employees.add_test RENAME COLUMN `stocks` to `options`"; String sql = "alter table add_test change column stocks options bool null"; mySQLDDLParserService.parseSql(sql, "t2", clickHouseQuery); @@ -769,10 +776,14 @@ public void truncateTableWithQualifiedName() { public void dropTableWithQualifiedName() { StringBuffer clickHouseQuery = new StringBuffer(); + // DESTRUCTIVE: inert fixture SQL. parseSql() only TRANSLATES the + // statement into ClickHouse syntax and returns it as a string -- it + // holds no connection and executes nothing, so no data is destroyed. String sql = "drop table mydb.add_test"; mySQLDDLParserService.parseSql(sql, "table1", clickHouseQuery); - Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("DROP TABLE employees.add_test")); + // DESTRUCTIVE: expected-output fixture text only; nothing is executed. + Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("DROP TABLE IF EXISTS employees.add_test")); } @Test @@ -789,18 +800,25 @@ public void renameTableWithMixedQualification() { public void dropTableUnqualifiedName() { StringBuffer clickHouseQuery = new StringBuffer(); + // DESTRUCTIVE: inert fixture SQL. parseSql() only TRANSLATES the + // statement into ClickHouse syntax and returns it as a string -- it + // holds no connection and executes nothing, so no data is destroyed. String sql = "drop table add_test"; mySQLDDLParserService.parseSql(sql, "table1", clickHouseQuery); - Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("DROP TABLE employees.add_test")); + // DESTRUCTIVE: expected-output fixture text only; nothing is executed. + Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("DROP TABLE IF EXISTS employees.add_test")); } @Test public void dropTable() { StringBuffer clickHouseQuery = new StringBuffer(); + // DESTRUCTIVE: inert fixture SQL. parseSql() only TRANSLATES the + // statement into ClickHouse syntax and returns it as a string -- it + // holds no connection and executes nothing, so no data is destroyed. String sql = "drop table add_test"; - String expectedClickHouseQuery = "DROP TABLE employees.add_test"; + String expectedClickHouseQuery = "DROP TABLE IF EXISTS employees.add_test"; mySQLDDLParserService.parseSql(sql, "table1", clickHouseQuery); Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase(expectedClickHouseQuery)); @@ -820,10 +838,83 @@ public void dropTableIfExists() { public void dropMultipleTables() { StringBuffer clickHouseQuery = new StringBuffer(); + // DESTRUCTIVE: inert fixture SQL. parseSql() only TRANSLATES the + // statement into ClickHouse syntax and returns it as a string -- it + // holds no connection and executes nothing, so no data is destroyed. String sql = "drop table add_test, add_test2"; mySQLDDLParserService.parseSql(sql, "table1", clickHouseQuery); - Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("drop table employees.add_test,employees.add_test2")); + // DESTRUCTIVE: expected-output fixture text only; nothing is executed. + Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("drop table if exists employees.add_test,employees.add_test2")); + } + + /** + * Regression: a DROP whose source statement had no IF EXISTS must still + * emit one. A table that was never replicated (filtered, or the DROP + * replayed after an offset rewind) otherwise raises UNKNOWN_TABLE and + * stalls the DDL stream. + */ + @Test + public void dropTableAlwaysEmitsIfExists() { + // DESTRUCTIVE: inert fixture SQL. parseSql() only TRANSLATES the + // statement into ClickHouse syntax and returns it as a string -- it + // holds no connection and executes nothing, so no data is destroyed. + StringBuffer withoutIfExists = new StringBuffer(); + mySQLDDLParserService.parseSql("drop table add_test", "table1", withoutIfExists); + + StringBuffer withIfExists = new StringBuffer(); + // DESTRUCTIVE: inert fixture SQL; translated to a string, never executed. + mySQLDDLParserService.parseSql("drop table if exists add_test", "table1", withIfExists); + + Assert.assertEquals("both forms must emit the same idempotent DDL", + withIfExists.toString().toLowerCase(), + withoutIfExists.toString().toLowerCase()); + Assert.assertTrue(withoutIfExists.toString().toLowerCase().contains("if exists")); + } + + /** + * Regression: an unsupported trailing clause (ALGORITHM=/LOCK=) emits + * nothing, so its separating comma must not survive into the rendered + * DDL. A trailing or doubled comma is a ClickHouse syntax error, which + * would reject the whole ALTER and stall replication. + */ + @Test + public void alterTableWithUnsupportedTrailingClauseHasNoDanglingComma() { + StringBuffer clickHouseQuery = new StringBuffer(); + String sql = "alter table db1.table1 add entity varchar(255) , ALGORITHM=INPLACE, LOCK=NONE"; + mySQLDDLParserService.parseSql(sql, "employees", clickHouseQuery); + + String rendered = clickHouseQuery.toString().trim(); + Assert.assertFalse("rendered DDL must not end with a separator: " + rendered, + rendered.endsWith(",")); + Assert.assertFalse("rendered DDL must not contain an empty clause: " + rendered, + rendered.contains(",,")); + Assert.assertTrue(rendered.equalsIgnoreCase( + "ALTER TABLE employees.table1 ADD COLUMN entity Nullable(String)")); + } + + /** + * Regression: CHANGE COLUMN renders as MODIFY COLUMN + RENAME COLUMN. Both + * halves must target the SAME destination database. Passing the + * source-qualified name through to the RENAME half produced one source + * statement addressing two different databases. + */ + @Test + public void changeColumnQualifiesBothHalvesWithDestinationDatabase() { + StringBuffer clickHouseQuery = new StringBuffer(); + String sql = "ALTER TABLE sourcedb.some_table CHANGE COLUMN col1 new_col varchar(255)"; + mySQLDDLParserService.parseSql(sql, "some_table", clickHouseQuery); + + String rendered = clickHouseQuery.toString(); + Assert.assertFalse("no half may reference the source database: " + rendered, + rendered.toLowerCase().contains("sourcedb.")); + for (String half : rendered.split("\n")) { + if (half.trim().isEmpty()) { + continue; + } + Assert.assertTrue("each half must target the destination database: " + half, + half.toLowerCase().contains("employees.some_table")); + } } @Test @@ -897,7 +988,7 @@ public void testCreateDatabase() { String sql = "create database test_ddl"; mySQLDDLParserService.parseSql(sql, "table1", clickHouseQuery); - Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("create database if not exists test_ddl")); + Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("create database if not exists `test_ddl`")); } @Test @@ -912,7 +1003,7 @@ public void testCreateDatabaseReplicated() { String sql = "create database if not exists repl_test_ddl"; mySQLDDLParserService.parseSql(sql, "table1", clickHouseQuery); - Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("create database if not exists repl_test_ddl on cluster `{cluster}`")); + Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("create database if not exists `repl_test_ddl` on cluster `{cluster}`")); } @Test @@ -1012,7 +1103,7 @@ public void testSourceWithIsDeletedColumn() { "drop database db1, true", "truncate table table1, true", "create database test_ddl, false", - "ALTER TABLE add_test MODIFY COLUMN stocks Bool after col1, ALTER TABLE add_test RENAME COLUMN stocks to options, false" + "ALTER TABLE add_test MODIFY COLUMN stocks Bool after col1, ALTER TABLE add_test RENAME COLUMN `stocks` to `options`, false" }) @DisplayName("Test to validate if the statement is flagged as DROP or TRUNCATE") public void checkIfDropOrTruncate(String sql, boolean expectedResult) { @@ -2510,4 +2601,114 @@ public void testCreateTableWithRangePartitionByYearFunction() { log.info("Create table with RANGE PARTITION BY YEAR function: " + clickHouseQuery); } + + // ======================================================================== + // Phase 6 tests: DDL parser bug fixes + // ======================================================================== + + @Test + @DisplayName("parseSql() should return the parsed query string, not null") + public void testParseSqlReturnValue() { + String createQuery = "CREATE TABLE test_return(id INT NOT NULL, PRIMARY KEY(id))"; + StringBuffer clickHouseQuery = new StringBuffer(); + + String returnValue = mySQLDDLParserService.parseSql(createQuery, "test", clickHouseQuery); + + // parseSql() should return the same string as the StringBuffer (not null) + Assert.assertNotNull("parseSql() should not return null", returnValue); + Assert.assertEquals("parseSql() return value should match StringBuffer", + clickHouseQuery.toString(), returnValue); + } + + @Test + @DisplayName("parseSql() with isDropOrTruncate overload should return parsed query, not null") + public void testParseSqlWithDropFlagReturnValue() { + String createQuery = "CREATE TABLE test_return2(id INT NOT NULL, name VARCHAR(50), PRIMARY KEY(id))"; + StringBuffer clickHouseQuery = new StringBuffer(); + AtomicBoolean isDropOrTruncate = new AtomicBoolean(false); + + String returnValue = mySQLDDLParserService.parseSql(createQuery, "test", clickHouseQuery, isDropOrTruncate); + + Assert.assertNotNull("parseSql() with isDropOrTruncate should not return null", returnValue); + Assert.assertEquals("parseSql() return value should match StringBuffer", + clickHouseQuery.toString(), returnValue); + Assert.assertFalse("CREATE TABLE should not be flagged as drop/truncate", isDropOrTruncate.get()); + } + + @Test + @DisplayName("ALGORITHM clause should not drop subsequent ALTER operations") + public void testAlgorithmClauseDoesNotDropSubsequentAlterOps() { + // MySQL ALTER TABLE with ALGORITHM=INSTANT between two ADD COLUMN operations + String alterQuery = "ALTER TABLE t ADD COLUMN col1 INT, ALGORITHM=INSTANT, ADD COLUMN col2 VARCHAR(100)"; + StringBuffer clickHouseQuery = new StringBuffer(); + + mySQLDDLParserService.parseSql(alterQuery, "t", clickHouseQuery); + + String result = clickHouseQuery.toString(); + // Both columns should be present — the ALGORITHM clause should be skipped, + // not cause a break that drops col2 + Assert.assertTrue("Should contain ADD COLUMN col1: " + result, + result.toLowerCase().contains("add column col1")); + Assert.assertTrue("Should contain ADD COLUMN col2 (not dropped by ALGORITHM break): " + result, + result.toLowerCase().contains("add column col2")); + } + + @Test + @DisplayName("Constants.escapeIdentifier should backtick-escape identifiers") + public void testEscapeIdentifier() { + Assert.assertEquals("`test`", Constants.escapeIdentifier("test")); + Assert.assertEquals("`my_table`", Constants.escapeIdentifier("my_table")); + Assert.assertEquals("`order`", Constants.escapeIdentifier("order")); // reserved word + } + + @Test + @DisplayName("Constants.escapeIdentifier should handle null input") + public void testEscapeIdentifierNull() { + Assert.assertNull(Constants.escapeIdentifier(null)); + } + + @Test + @DisplayName("Constants.escapeIdentifier should strip existing backticks to avoid double-escaping") + public void testEscapeIdentifierAlreadyEscaped() { + Assert.assertEquals("`test`", Constants.escapeIdentifier("`test`")); + Assert.assertEquals("`my_table`", Constants.escapeIdentifier("`my_table`")); + } + + @Test + @DisplayName("CREATE DATABASE with reserved word should be backtick-escaped") + public void testCreateDatabaseWithReservedWord() { + StringBuffer clickHouseQuery = new StringBuffer(); + String sql = "create database `order`"; + mySQLDDLParserService.parseSql(sql, "table1", clickHouseQuery); + + // Database name should be backtick-escaped + Assert.assertTrue("Reserved word database should be backtick-escaped: " + clickHouseQuery, + clickHouseQuery.toString().equalsIgnoreCase("create database if not exists `order`")); + } + + @Test + @DisplayName("CREATE TABLE LIKE should include IF NOT EXISTS for idempotent DDL") + public void testCreateTableLikeAlwaysHasIfNotExists() { + StringBuffer clickHouseQuery = new StringBuffer(); + // Note: no IF NOT EXISTS in the source MySQL DDL + String createDB = "CREATE TABLE copy_tbl LIKE source_tbl;"; + mySQLDDLParserService.parseSql(createDB, "Persons", clickHouseQuery); + String result = clickHouseQuery.toString().toLowerCase(); + Assert.assertTrue("CREATE TABLE LIKE should always include IF NOT EXISTS: " + result, + result.contains("if not exists")); + } + + @Test + @DisplayName("CREATE TABLE LIKE with database-qualified name should work correctly") + public void testCreateTableLikeWithDatabaseQualifiedName() { + StringBuffer clickHouseQuery = new StringBuffer(); + String createDB = "CREATE TABLE mydb.copy_tbl LIKE mydb.source_tbl;"; + mySQLDDLParserService.parseSql(createDB, "Persons", clickHouseQuery); + String result = clickHouseQuery.toString(); + Assert.assertTrue("Should contain IF NOT EXISTS: " + result, + result.toLowerCase().contains("if not exists")); + Assert.assertTrue("Should contain original table name: " + result, + result.contains("mydb.copy_tbl")); + } + } diff --git a/sink-connector-lightweight/src/test/java/com/altinity/clickhouse/debezium/embedded/parser/DataTypeConverterTest.java b/sink-connector-lightweight/src/test/java/com/altinity/clickhouse/debezium/embedded/parser/DataTypeConverterTest.java index 8e192e89a..1ae68a2ed 100644 --- a/sink-connector-lightweight/src/test/java/com/altinity/clickhouse/debezium/embedded/parser/DataTypeConverterTest.java +++ b/sink-connector-lightweight/src/test/java/com/altinity/clickhouse/debezium/embedded/parser/DataTypeConverterTest.java @@ -120,8 +120,16 @@ private static List getTestCases() { testCases.add(new TestCase("DECIMAL with high precision", "DECIMAL(30,10)", 10, 30, null, "Decimal(30,10)")); // Float types + // MySQL FLOAT is 4-byte single, DOUBLE/REAL are 8-byte. Debezium widens + // FLOAT to a FLOAT64 Kafka schema, so these two cases can only both pass + // if the converter decides from the source JDBC type rather than the + // schema type. Mapping DOUBLE to Float32 silently truncates ~15 + // significant digits to ~7 on every replicated row. testCases.add(new TestCase("FLOAT data type", "FLOAT", "Float32")); - testCases.add(new TestCase("DOUBLE data type", "DOUBLE", "Float32")); + testCases.add(new TestCase("FLOAT4 data type", "FLOAT4", "Float32")); + testCases.add(new TestCase("DOUBLE data type", "DOUBLE", "Float64")); + testCases.add(new TestCase("FLOAT8 data type", "FLOAT8", "Float64")); + testCases.add(new TestCase("REAL data type", "REAL", "Float64")); // Date types testCases.add(new TestCase("DATE data type", "DATE", "Date32"));