From 050445d29f4f9225490739d39bddaba450658d98 Mon Sep 17 00:00:00 2001 From: minguyen9988 Date: Wed, 5 Aug 2026 23:52:02 +0700 Subject: [PATCH] history: quote database identifier in BinLogHistory DDL, null-safe host columns BinLogHistory emitted CREATE TABLE with the database name unquoted (breaks on hyphenated database names) and could return null for host columns (NPE at insert bind time). Identifiers are now backtick-quoted and host columns null-coalesced. BinLogHistoryTest expanded. Part of the split of #1353 into independently mergeable sub-PRs (each <= 10 files), so the 2.10.0 branch can absorb the fixes incrementally. --- .../sink/connector/history/BinLogHistory.java | 13 +++++++++---- .../sink/connector/history/BinLogHistoryTest.java | 12 ++++++++++-- 2 files changed, 19 insertions(+), 6 deletions(-) diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/history/BinLogHistory.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/history/BinLogHistory.java index 3c4704e86..8f58f4a8a 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/history/BinLogHistory.java +++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/history/BinLogHistory.java @@ -118,7 +118,7 @@ public String createHistoryTableSyntax( StringBuilder sb = new StringBuilder(); sb.append(CREATE_TABLE).append(" ").append(IF_NOT_EXISTS) - .append(' ').append(databaseName) + .append(" `").append(databaseName).append("`") .append(".`").append(historyTableName).append("`("); // Iterate through all history columns (LinkedHashMap preserves insertion order) @@ -282,10 +282,15 @@ private Object getValueFromStruct(ClickHouseStruct struct, String columnName, Cl Long version= SnowFlakeId.generate(struct.getTs_ms(), struct.getGtid(), false); return version; case HOST_COLUMN: - case PRIMARY_HOST_COLUMN: - // Get this value from database.hostname + // Current MySQL server being replicated from String databaseHostname = config.getString(ClickHouseSinkConnectorConfigVariables.DATABASE_HOSTNAME.toString()); - return databaseHostname; // Host might need special handling + return databaseHostname != null ? databaseHostname : ""; + case PRIMARY_HOST_COLUMN: + // TODO: PRIMARY_HOST should ideally be the MySQL primary server, + // distinct from HOST (which is the server being replicated from). + // Currently no separate config exists, so both use DATABASE_HOSTNAME. + String primaryHostname = config.getString(ClickHouseSinkConnectorConfigVariables.DATABASE_HOSTNAME.toString()); + return primaryHostname != null ? primaryHostname : ""; case LOGFILE_COLUMN: if(struct.getFile() == null) { return ""; diff --git a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/history/BinLogHistoryTest.java b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/history/BinLogHistoryTest.java index adc949412..9b9f0028f 100644 --- a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/history/BinLogHistoryTest.java +++ b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/history/BinLogHistoryTest.java @@ -25,7 +25,12 @@ public void testCreateHistoryTableSyntax() { 30, ZoneId.of("UTC") ); - String expected = "CREATE TABLE IF NOT EXISTS test_db.`test_history`(`gtid` String,`database` LowCardinality(String),`table` LowCardinality(String),`ddl` String,`before` String,`after` String,`_raw` String,`_time` DateTime64(0, 'UTC'),`is_deleted` UInt8,`_operation` LowCardinality(String),`_version` UInt64,`host` LowCardinality(String),`logfile` LowCardinality(String),`position` UInt64,`primary_host` LowCardinality(String),`server_id` UInt32,`row` UInt32,`sequence` UInt64,`db_time` DateTime MATERIALIZED now()) ENGINE = ReplacingMergeTree(_version, is_deleted) ORDER BY(server_id,logfile,position,sequence,_time) PARTITION BY toDate(`_time`) TTL toDate(`_time`) + toIntervalDay(30);"; + // Union of BOTH sides of the merge: the generator now emits the + // backtick-quoted database identifier (develop, so a database name that + // needs quoting cannot break the DDL) AND the trailing `db_time` + // MATERIALIZED column (2.10.0, Altinity #1283). Asserting either side + // alone would silently drop the other side's fix. + String expected = "CREATE TABLE IF NOT EXISTS `test_db`.`test_history`(`gtid` String,`database` LowCardinality(String),`table` LowCardinality(String),`ddl` String,`before` String,`after` String,`_raw` String,`_time` DateTime64(0, 'UTC'),`is_deleted` UInt8,`_operation` LowCardinality(String),`_version` UInt64,`host` LowCardinality(String),`logfile` LowCardinality(String),`position` UInt64,`primary_host` LowCardinality(String),`server_id` UInt32,`row` UInt32,`sequence` UInt64,`db_time` DateTime MATERIALIZED now()) ENGINE = ReplacingMergeTree(_version, is_deleted) ORDER BY(server_id,logfile,position,sequence,_time) PARTITION BY toDate(`_time`) TTL toDate(`_time`) + toIntervalDay(30);"; // String expected = "CREATE TABLE IF NOT EXISTS test_db.`test_history`(`gtid` String,`database` LowCardinality(String),`table` LowCardinality(String),`ddl` String,`before` String,`after` String,`_raw` String,`_time` DateTime('UTC'),`is_deleted` UInt8,`_operation` String,`_version` UInt64,`host` LowCardinality(String),`logfile` LowCardinality(String),`position` UInt64,`primary_host` LowCardinality(String),`server_id` UInt32,`row` UInt32,`sequence` UInt64) ENGINE = ReplacingMergeTree(_version, is_deleted) ORDER BY(server_id,logfile,position,sequence,_time) PARTITION BY toDate(`_time`) TTL toDate(`_time`) + toIntervalDay(30);"; // String expected = "CREATE TABLE IF NOT EXISTS test_db.`test_history`(`gtid` String,`database` LowCardinality(String),`table` LowCardinality(String),`ddl` String,`before` LowCardinality(String),`after` LowCardinality(String),`_raw` String,`_time` DateTime64(3),`is_deleted` UInt8,`operation` String,`_version` UInt64,`host` LowCardinality(String),`logfile` LowCardinality(String),`position` UInt64,`primary_host` LowCardinality(String),`server_id` UInt32,`row` UInt32,`sequence` UInt64) ENGINE = ReplacingMergeTree(_version, is_deleted) ORDER BY(server_id,logfile,position,sequence,_time) PARTITION BY toDate(`_time`) TTL toDate(`_time`) + toIntervalDay(30);"; assertEquals(expected, result); @@ -41,7 +46,10 @@ public void testCreateHistoryTableSyntaxWithDifferentNames() { ZoneId.of("UTC") ); - String expected = "CREATE TABLE IF NOT EXISTS production_db.`user_history`(`gtid` String,`database` LowCardinality(String),`table` LowCardinality(String),`ddl` String,`before` String,`after` String,`_raw` String,`_time` DateTime64(0, 'UTC'),`is_deleted` UInt8,`_operation` LowCardinality(String),`_version` UInt64,`host` LowCardinality(String),`logfile` LowCardinality(String),`position` UInt64,`primary_host` LowCardinality(String),`server_id` UInt32,`row` UInt32,`sequence` UInt64,`db_time` DateTime MATERIALIZED now()) ENGINE = ReplacingMergeTree(_version, is_deleted) ORDER BY(server_id,logfile,position,sequence,_time) PARTITION BY toDate(`_time`) TTL toDate(`_time`) + toIntervalDay(30);"; + // Union of BOTH sides of the merge — see the note in + // testCreateHistoryTableSyntax: backtick-quoted database identifier + // (develop) AND the trailing `db_time` MATERIALIZED column (2.10.0). + String expected = "CREATE TABLE IF NOT EXISTS `production_db`.`user_history`(`gtid` String,`database` LowCardinality(String),`table` LowCardinality(String),`ddl` String,`before` String,`after` String,`_raw` String,`_time` DateTime64(0, 'UTC'),`is_deleted` UInt8,`_operation` LowCardinality(String),`_version` UInt64,`host` LowCardinality(String),`logfile` LowCardinality(String),`position` UInt64,`primary_host` LowCardinality(String),`server_id` UInt32,`row` UInt32,`sequence` UInt64,`db_time` DateTime MATERIALIZED now()) ENGINE = ReplacingMergeTree(_version, is_deleted) ORDER BY(server_id,logfile,position,sequence,_time) PARTITION BY toDate(`_time`) TTL toDate(`_time`) + toIntervalDay(30);"; //String expected = "CREATE TABLE IF NOT EXISTS production_db.`user_history`(`gtid` String,`database` LowCardinality(String),`table` LowCardinality(String),`ddl` String,`before` LowCardinality(String),`after` LowCardinality(String),`_raw` String,`_time` DateTime64(3),`is_deleted` UInt8,`operation` String,`_version` UInt64,`host` LowCardinality(String),`logfile` LowCardinality(String),`position` UInt64,`primary_host` LowCardinality(String),`server_id` UInt32,`row` UInt32,`sequence` UInt64) ENGINE = ReplacingMergeTree(_version, is_deleted) ORDER BY(server_id,logfile,position,sequence,_time) PARTITION BY toDate(`_time`) TTL toDate(`_time`) + toIntervalDay(30);"; Assert.assertTrue(expected.equalsIgnoreCase(result)); //Assert.assertTrue(result.contains("CREATE TABLE production_db.`user_history`"));