diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/ClickHouseDbConstants.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/ClickHouseDbConstants.java index 066e65ef0..ce3f38bed 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/ClickHouseDbConstants.java +++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/ClickHouseDbConstants.java @@ -32,7 +32,7 @@ public class ClickHouseDbConstants { /** * The CREATE TABLE statement keyword. */ - public static final String CREATE_TABLE = "CREATE TABLE"; + public static final String CREATE_TABLE = "CREATE TABLE IF NOT EXISTS"; /** * Represents a nullability specification of NULL. @@ -218,9 +218,29 @@ public class ClickHouseDbConstants { public static final String PRIMARY_HOST_COLUMN_DATA_TYPE = "String"; /** - * A SQL statement used to create the topic_offset_metadata - * table for managing Kafka offsets. + * Default table name for the Kafka offset metadata table. */ + public static final String DEFAULT_OFFSET_TABLE_NAME = "topic_offset_metadata"; + + /** + * Returns a SQL statement to create the offset metadata table with + * the specified table name. + * + * @param tableName the name of the offset table to create + * @return the CREATE TABLE SQL statement + */ + public static String getOffsetTableCreateSql(String tableName) { + return "CREATE TABLE " + tableName + "(`_topic` String, " + + "`_partition` UInt64,`_offset` SimpleAggregateFunction(max, " + + "UInt64))ENGINE = AggregatingMergeTree ORDER BY " + + "(_topic, _partition)"; + } + + /** + * Returns a SQL statement to create the default offset metadata table. + * @deprecated Use {@link #getOffsetTableCreateSql(String)} with a configured table name. + */ + @Deprecated public static final String OFFSET_TABLE_CREATE_SQL = "CREATE TABLE topic_offset_metadata(`_topic` String, " + "`_partition` UInt64,`_offset` SimpleAggregateFunction(max, " diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/DbKafkaOffsetWriter.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/DbKafkaOffsetWriter.java index 237e2e1b8..b36167e26 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/DbKafkaOffsetWriter.java +++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/DbKafkaOffsetWriter.java @@ -22,6 +22,11 @@ public class DbKafkaOffsetWriter extends BaseDbWriter { */ String query; + /** + * The name of the offset table (stored for use in getStoredOffsets). + */ + String tableName; + /** * A map of column names to their respective data types in the offset table. */ @@ -61,6 +66,7 @@ public DbKafkaOffsetWriter( super(hostName, port, database, userName, password, config, connection); + this.tableName = tableName; createOffsetTable(); this.columnNamesToDataTypesMap = new DBMetadata(config).getColumnsDataTypesForTable( @@ -77,13 +83,11 @@ public DbKafkaOffsetWriter( * Function to create the Kafka offset table if it does not exist. */ public void createOffsetTable() { - try { - PreparedStatement ps = this.getConnection().prepareStatement( - ClickHouseDbConstants.OFFSET_TABLE_CREATE_SQL - ); + try (PreparedStatement ps = this.getConnection().prepareStatement( + ClickHouseDbConstants.OFFSET_TABLE_CREATE_SQL)) { ps.execute(); } catch (SQLException se) { - log.error("Error creating Kafka offset table"); + log.error("Error creating Kafka offset table", se); } } @@ -148,22 +152,23 @@ public void insertTopicOffsetMetadata( public Map getStoredOffsets() throws SQLException { Map result = new HashMap<>(); - Statement stmt = this.getConnection().createStatement(); - ResultSet rs = stmt.executeQuery("select * from topic_offset_metadata"); - - while (rs.next()) { - String topicName = rs.getString( - KafkaMetaData.TOPIC.getColumn() - ); - int partition = rs.getInt( - KafkaMetaData.PARTITION.getColumn() - ); - long offset = rs.getLong( - KafkaMetaData.OFFSET.getColumn() - ); - - TopicPartition tp = new TopicPartition(topicName, partition); - result.put(tp, offset); + try (Statement stmt = this.getConnection().createStatement(); + ResultSet rs = stmt.executeQuery("select * from " + this.tableName)) { + + while (rs.next()) { + String topicName = rs.getString( + KafkaMetaData.TOPIC.getColumn() + ); + int partition = rs.getInt( + KafkaMetaData.PARTITION.getColumn() + ); + long offset = rs.getLong( + KafkaMetaData.OFFSET.getColumn() + ); + + TopicPartition tp = new TopicPartition(topicName, partition); + result.put(tp, offset); + } } return result; diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/QueryFormatter.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/QueryFormatter.java index e3ed1e348..17f0bc7c1 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/QueryFormatter.java +++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/QueryFormatter.java @@ -79,10 +79,35 @@ public MutablePair> getInsertQueryUsingInputFunctio boolean includeKafkaMetaData, boolean includeRawData, String rawDataColumn, String dbName) { + return getInsertQueryUsingInputFunction(tableName, fields, + columnNameToDataTypeMap, includeKafkaMetaData, includeRawData, + rawDataColumn, dbName, null); + } + + /** + * Variant of {@link #getInsertQueryUsingInputFunction(String, List, Map, + * boolean, boolean, String, String)} that also receives the + * connector-managed column names for this table (version/sign/is_deleted + * columns as actually named in the table engine definition, which may be + * custom). These columns are always kept in the insert column list even + * though they never appear in the source event — the field mapper computes + * their values. + * + * @param connectorManagedColumns writer-specific managed column names; may + * be null when unknown. + */ + public MutablePair> getInsertQueryUsingInputFunction( + String tableName, List fields, + Map columnNameToDataTypeMap, + boolean includeKafkaMetaData, + boolean includeRawData, + String rawDataColumn, String dbName, + java.util.Collection connectorManagedColumns) { // Create column data structures ColumnData columnData = createColumns(tableName, fields, columnNameToDataTypeMap, - includeKafkaMetaData, includeRawData, rawDataColumn, dbName); + includeKafkaMetaData, includeRawData, rawDataColumn, dbName, + connectorManagedColumns); if (columnData == null) { return null; @@ -117,6 +142,24 @@ private static class ColumnData { } } + /** + * Checks if a column is populated by the connector itself rather than by + * the source event: the ReplacingMergeTree bookkeeping columns + * ({@code _version}, {@code _sign}, {@code is_deleted}) and the SCD2 + * temporal tracking columns. These are never present in the source schema + * but MUST stay in the insert column list — the field mapper computes + * their values. + * + * @param colName the name of the column to check. + * @return true if the connector supplies this column's value itself. + */ + private boolean isConnectorManagedColumn(String colName) { + return colName.equalsIgnoreCase(ClickHouseDbConstants.VERSION_COLUMN) + || colName.equalsIgnoreCase(ClickHouseDbConstants.SIGN_COLUMN) + || colName.equalsIgnoreCase(ClickHouseDbConstants.IS_DELETED_COLUMN) + || isTemporalTrackingColumn(colName); + } + /** * Checks if a column is a temporal tracking column used for history. * These columns should use DEFAULT values from the table schema. @@ -171,20 +214,29 @@ private String formatParameterPlaceholder(String dataType) { * @return the precision value as a string, defaults to "3" if not found */ private String extractDateTime64Precision(String dataType) { - // Find the opening parenthesis - int start = dataType.lastIndexOf('('); + // Strip Nullable wrapper if present — e.g. "Nullable(DateTime64(3))" + String dt = dataType; + if (dt.toUpperCase().startsWith("NULLABLE(") && dt.endsWith(")")) { + dt = dt.substring(9, dt.length() - 1); + } + // Find the opening parenthesis. This MUST index into `dt` (the + // Nullable-stripped string), not the original `dataType`: every + // substring/indexOf below operates on `dt`, so mixing an index taken + // from `dataType` with a substring of `dt` reads the wrong characters + // for any Nullable(...) type. + int start = dt.indexOf('('); if (start == -1) { return "3"; // Default precision } // Find the first comma or closing parenthesis - int end = dataType.indexOf(',', start); + int end = dt.indexOf(',', start); if (end == -1) { - end = dataType.indexOf(')', start); + end = dt.indexOf(')', start); } if (end == -1 || end <= start + 1) { return "3"; // Default precision } - return dataType.substring(start + 1, end).trim(); + return dt.substring(start + 1, end).trim(); } /** @@ -237,7 +289,12 @@ private String formatLiteralForSql(Object value, String dataType) { numLiteral = numLiteral.replace("'", "''"); return "CAST('" + numLiteral + "', '" + dataType + "')"; } - if (upperDataType.contains("DATETIME64") || upperDataType.contains("DATETIME")) { + if (upperDataType.contains("DATETIME64")) { + String ts = value.toString().replace("'", "''"); + String precision = extractDateTime64Precision(dataType); + return "toDateTime64('" + ts + "', " + precision + ")"; + } + if (upperDataType.contains("DATETIME")) { String ts = value.toString().replace("'", "''"); return "toDateTime('" + ts + "')"; } @@ -270,7 +327,8 @@ private String formatLiteralForSql(Object value, String dataType) { * @return a ColumnData object containing the column index map and delimited strings, or null if fields is null. */ private ColumnData createColumns(String tableName, List fields, Map columnNameToDataTypeMap, - boolean includeKafkaMetaData, boolean includeRawData, String rawDataColumn, String dbName) { + boolean includeKafkaMetaData, boolean includeRawData, String rawDataColumn, String dbName, + java.util.Collection connectorManagedColumns) { if (fields == null) { log.error("getInsertQueryUsingInputFunction, fields empty"); @@ -283,6 +341,26 @@ private ColumnData createColumns(String tableName, List fields, Map sourceFieldNamesLower = new java.util.HashSet<>(); + for (Field f : fields) { + if (f != null && f.name() != null) { + sourceFieldNamesLower.add(f.name().toLowerCase(java.util.Locale.ROOT)); + } + } + + // Writer-specific connector-managed column names (custom version/sign + // columns parsed from the table engine), lower-cased. + java.util.Set managedLower = new java.util.HashSet<>(); + if (connectorManagedColumns != null) { + for (String c : connectorManagedColumns) { + if (c != null) { + managedLower.add(c.toLowerCase(java.util.Locale.ROOT)); + } + } + } + // Loop over each column to generate the insert query and map data types for (Map.Entry entry : columnNameToDataTypeMap.entrySet()) { String sourceColumnName = entry.getKey(); @@ -308,6 +386,25 @@ private ColumnData createColumns(String tableName, List fields, Map colNameToDataTypesMap, ALTER_TABLE_OPERATION operation) { + return createAlterTableSyntax(null, tableName, colNameToDataTypesMap, operation); + } + + /** + * Creates the SQL syntax for an ALTER TABLE operation with database prefix. + * + * @param databaseName the name of the database (may be null for backwards compatibility) + * @param tableName the name of the table to alter + * @param colNameToDataTypesMap a map of column names to data types + * @param operation the ALTER_TABLE_OPERATION to perform (ADD or REMOVE) + * @return a SQL string for altering the table + */ + public String createAlterTableSyntax(String databaseName, String tableName, + Map colNameToDataTypesMap, + ALTER_TABLE_OPERATION operation) { + + if (colNameToDataTypesMap == null || colNameToDataTypesMap.isEmpty()) { + log.warn("createAlterTableSyntax called with empty column map for table {}", tableName); + return ""; + } StringBuilder alterTableSyntax = new StringBuilder(); - alterTableSyntax.append(ClickHouseDbConstants.ALTER_TABLE) - .append(" ").append(tableName).append(" "); + alterTableSyntax.append(ClickHouseDbConstants.ALTER_TABLE).append(" "); + if (databaseName != null && !databaseName.isEmpty()) { + alterTableSyntax.append("`").append(databaseName).append("`.`").append(tableName).append("`"); + } else { + alterTableSyntax.append("`").append(tableName).append("`"); + } + alterTableSyntax.append(" "); for (Map.Entry entry : colNameToDataTypesMap.entrySet()) { diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseAutoCreateTable.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseAutoCreateTable.java index c20b281bf..d48f598d8 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseAutoCreateTable.java +++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseAutoCreateTable.java @@ -124,7 +124,7 @@ public String createTableSyntax(ArrayList primaryKey, StringBuilder createTableSyntax = new StringBuilder(); createTableSyntax.append(CREATE_TABLE).append(" ") - .append(databaseName).append(".") + .append("`").append(databaseName).append("`").append(".") .append("`").append(tableName).append("`"); if (useReplicatedReplacingMergeTree == true) { createTableSyntax.append(" ON CLUSTER `{cluster}` "); @@ -249,12 +249,12 @@ public String createTableSyntax(ArrayList primaryKey, && isPrimaryKeyColumnPresent(primaryKey, columnToDataTypesMap)) { createTableSyntax.append(PRIMARY_KEY).append("("); createTableSyntax.append(primaryKey.stream() - .map(Object::toString) + .map(pk -> "`" + pk + "`") .collect(Collectors.joining(","))); createTableSyntax.append(") "); createTableSyntax.append(ORDER_BY).append("("); createTableSyntax.append(primaryKey.stream() - .map(Object::toString) + .map(pk -> "`" + pk + "`") .collect(Collectors.joining(","))); if(config.getBoolean(ClickHouseSinkConnectorConfigVariables.REPLICATION_HISTORY_ENABLE.toString())) { createTableSyntax.append(",`").append(DELETED_TIME_COLUMN).append("`"); @@ -339,7 +339,7 @@ private ZoneId getServerTimeZone(ClickHouseSinkConnectorConfig config, Connectio } public void createHistoryDatabase(String databaseName, Connection connection, ClickHouseSinkConnectorConfig config) throws SQLException { - String sql = "CREATE DATABASE IF NOT EXISTS " + databaseName; + String sql = "CREATE DATABASE IF NOT EXISTS `" + databaseName.replace("`", "``") + "`"; log.info(String.format( "**** AUTO CREATE HISTORY DATABASE for database(%s), Query :%s)", databaseName, sql)); diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseCreateDatabase.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseCreateDatabase.java index 8aa43050b..77503c0e9 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseCreateDatabase.java +++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseCreateDatabase.java @@ -25,7 +25,9 @@ public class ClickHouseCreateDatabase extends ClickHouseTableOperationsBase { public void createNewDatabase(Connection conn, String dbName, Boolean useOnCluster, ClickHouseSinkConnectorConfig config) throws SQLException { String onCluster = useOnCluster ? " ON CLUSTER `{cluster}`" : ""; - String query = String.format("CREATE DATABASE IF NOT EXISTS %s%s", dbName, onCluster); + // Backtick-escape database name to handle reserved words and special characters + String escapedDbName = "`" + dbName.replace("`", "") + "`"; + String query = String.format("CREATE DATABASE IF NOT EXISTS %s%s", escapedDbName, onCluster); DBMetadata metadata = new DBMetadata(config); metadata.executeSystemQuery(conn, query); } diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseTableOperationsBase.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseTableOperationsBase.java index 3f5b3d5d8..9121960b2 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseTableOperationsBase.java +++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseTableOperationsBase.java @@ -12,7 +12,7 @@ import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; -import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.Map; import static com.altinity.clickhouse.sink.connector.config.DefaultColumnDataTypeMappingConfig.loadDefaultColumnDataTypeMapping; @@ -89,7 +89,7 @@ public ClickHouseTableOperationsBase() { */ public Map getColumnNameToCHDataTypeMapping(Field[] fields, ClickHouseSinkConnectorConfig config) { ClickHouseDataTypeMapper mapper = new ClickHouseDataTypeMapper(); - Map columnToDataTypesMap = new HashMap<>(); + Map columnToDataTypesMap = new LinkedHashMap<>(); for (Field f : fields) { String colName = f.name(); diff --git a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/QueryFormatterTest.java b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/QueryFormatterTest.java index 69f6cdc94..ef4b53479 100644 --- a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/QueryFormatterTest.java +++ b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/QueryFormatterTest.java @@ -6,6 +6,7 @@ import org.apache.kafka.connect.data.Schema; import org.junit.Assert; import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import java.util.ArrayList; @@ -35,6 +36,54 @@ public static void initialize() { fields.add(new Field("Min Value", 6, Schema.INT32_SCHEMA)); fields.add(new Field("Null Value", 7, Schema.INT32_SCHEMA)); } + @Test + public void testDestinationOnlyDataColumnIsOmittedFromInsert() { + // ADD COLUMN scenario: destination has price_usd but the (pre-ALTER) + // source event does not. The column must be OMITTED from the insert so + // ClickHouse fills its DEFAULT -- binding NULL fails outright on + // non-nullable columns and the whole batch then errors and replays + // forever (the price_usd data-loss/stall incident). + QueryFormatter qf = new QueryFormatter(); + + Map colMap = new HashMap<>(); + colMap.put("customerName", "String"); + colMap.put("price_usd", "Decimal(18, 6)"); // destination-only + colMap.put("_version", "UInt64"); // connector-managed: kept + + MutablePair> response = + qf.getInsertQueryUsingInputFunction("products", fields, colMap, + false, false, null, "employees"); + + Assert.assertFalse("Destination-only data column must be omitted", + response.left.contains("price_usd")); + Assert.assertTrue("Connector-managed _version must be kept", + response.left.contains("`_version`")); + Assert.assertTrue("Source-present column must be kept", + response.left.contains("`customerName`")); + Assert.assertFalse(response.right.containsKey("price_usd")); + } + + @Test + public void testCustomManagedColumnsAreKept() { + // Custom version/sign column names (e.g. ReplacingMergeTree(ver)) are + // passed through the connectorManagedColumns parameter and must be + // kept even though the source event never carries them. + QueryFormatter qf = new QueryFormatter(); + + Map colMap = new HashMap<>(); + colMap.put("customerName", "String"); + colMap.put("ver", "UInt64"); // custom version column + colMap.put("signv", "Int8"); // custom delete column + + MutablePair> response = + qf.getInsertQueryUsingInputFunction("products", fields, colMap, + false, false, null, "employees", + java.util.List.of("ver", "signv")); + + Assert.assertTrue(response.left.contains("`ver`")); + Assert.assertTrue(response.left.contains("`signv`")); + } + @Test public void testGetInsertQueryUsingInputFunctionWithKafkaMetaDataEnabled() { QueryFormatter qf = new QueryFormatter(); @@ -380,4 +429,89 @@ public void testGetInsertQueryForDeleteWithStringPrimaryKey() { Assert.assertTrue("Column index map should be empty", columnIndexMap.isEmpty()); } + + @Test + @DisplayName("extractDateTime64Precision handles Nullable(DateTime64(6)) wrapper") + public void testExtractDateTime64PrecisionWithNullable() { + // Test via reflection since extractDateTime64Precision is private + QueryFormatter qf = new QueryFormatter(); + try { + java.lang.reflect.Method method = QueryFormatter.class.getDeclaredMethod( + "extractDateTime64Precision", String.class); + method.setAccessible(true); + + // Plain DateTime64(3) + String result1 = (String) method.invoke(qf, "DateTime64(3)"); + Assert.assertEquals("Should extract precision 3", "3", result1); + + // Nullable(DateTime64(6)) + String result2 = (String) method.invoke(qf, "Nullable(DateTime64(6))"); + Assert.assertEquals("Should extract precision 6 from Nullable wrapper", "6", result2); + + // Nullable(DateTime64(9, 'UTC')) + String result3 = (String) method.invoke(qf, "Nullable(DateTime64(9, 'UTC'))"); + Assert.assertEquals("Should extract precision 9 from Nullable with timezone", "9", result3); + + // DateTime64 without precision + String result4 = (String) method.invoke(qf, "DateTime64"); + Assert.assertEquals("Should default to 3 when no precision specified", "3", result4); + + // Nested Nullable edge case + String result5 = (String) method.invoke(qf, "NULLABLE(DateTime64(4))"); + Assert.assertEquals("Should handle uppercase NULLABLE", "4", result5); + + } catch (Exception e) { + Assert.fail("Reflection failed: " + e.getMessage()); + } + } + + + @Test + @DisplayName("formatLiteralForSql uses toDateTime64 for DateTime64 types with correct precision") + public void testFormatLiteralForSqlDateTime64VsDateTime() { + QueryFormatter qf = new QueryFormatter(); + try { + // Actual signature is formatLiteralForSql(Object value, String dataType). + // The column name and timezone are not parameters of this helper — the + // timezone is applied by the callers that build toDateTime(...) expressions + // (see getInsertQueryForDelete / getInsertQueryForUpdate). + java.lang.reflect.Method method = QueryFormatter.class.getDeclaredMethod( + "formatLiteralForSql", Object.class, String.class); + method.setAccessible(true); + + // DateTime should use toDateTime + String dtResult = (String) method.invoke(qf, "2025-01-01 00:00:00", "DateTime"); + Assert.assertTrue("DateTime should use toDateTime function", + dtResult.contains("toDateTime(")); + Assert.assertFalse("DateTime should NOT use toDateTime64", + dtResult.contains("toDateTime64(")); + + // DateTime64(3) should use toDateTime64 with precision 3 + String dt64Result = (String) method.invoke(qf, "2025-01-01 00:00:00.000", "DateTime64(3)"); + Assert.assertTrue("DateTime64 should use toDateTime64 function", + dt64Result.contains("toDateTime64(")); + Assert.assertTrue("DateTime64(3) should include precision 3", + dt64Result.contains(", 3)")); + + // DateTime64(6) should use toDateTime64 with precision 6 + String dt64_6Result = (String) method.invoke(qf, "2025-01-01 00:00:00.000000", "DateTime64(6)"); + Assert.assertTrue("DateTime64(6) should use toDateTime64 function", + dt64_6Result.contains("toDateTime64(")); + Assert.assertTrue("DateTime64(6) should include precision 6", + dt64_6Result.contains(", 6)")); + + // Nullable(DateTime64(3)) should also use toDateTime64, with the + // Nullable wrapper stripped before precision extraction. + String nullableDt64 = (String) method.invoke(qf, "2025-01-01 00:00:00.000", + "Nullable(DateTime64(3))"); + Assert.assertTrue("Nullable(DateTime64) should use toDateTime64", + nullableDt64.contains("toDateTime64(")); + Assert.assertTrue("Nullable(DateTime64(3)) should include precision 3", + nullableDt64.contains(", 3)")); + + } catch (Exception e) { + Assert.fail("Reflection failed: " + e.getMessage()); + } + } + } diff --git a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseAlterTableTest.java b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseAlterTableTest.java index dfe4047fe..6bcdecd56 100644 --- a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseAlterTableTest.java +++ b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseAlterTableTest.java @@ -1,27 +1,84 @@ package com.altinity.clickhouse.sink.connector.db.operations; -import org.junit.Assert; +import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; -public class ClickHouseAlterTableTest extends com.altinity.clickhouse.sink.connector.db.operations.ClickHouseAutoCreateTableTest { +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; - @Test - public void createAlterTableSyntaxTest() { +import static org.junit.jupiter.api.Assertions.*; - ClickHouseAlterTable cat = new ClickHouseAlterTable(); +/** + * Tests for {@link ClickHouseAlterTable} SQL syntax generation. + */ +public class ClickHouseAlterTableTest { - // Add Column - String alterTableAddColumnQuery = cat.createAlterTableSyntax("employees", - this.getExpectedColumnToDataTypesMap(), ClickHouseAlterTable.ALTER_TABLE_OPERATION.ADD); + @Test + @DisplayName("Empty column map returns empty string") + public void testEmptyColumnMapReturnsEmpty() { + ClickHouseAlterTable alterTable = new ClickHouseAlterTable(); + String result = alterTable.createAlterTableSyntax( + "test_db", "test_table", + Collections.emptyMap(), + ClickHouseAlterTable.ALTER_TABLE_OPERATION.ADD); + assertEquals("", result, "Empty column map should return empty string"); + } - String expectedAddColumnQuery = "ALTER TABLE employees add column `amount` Float64,add column `occupation` String,add column `quantity` Int32,add column `blob_storage_scale` Decimal,add column `json_output` JSON,add column `max_amount` Float64,add column `amount_1` Float32,add column `customerName` String,add column `blob_storage` String,add column `employed` Bool"; - Assert.assertTrue(alterTableAddColumnQuery.equalsIgnoreCase(expectedAddColumnQuery)); + @Test + @DisplayName("Null column map returns empty string") + public void testNullColumnMapReturnsEmpty() { + ClickHouseAlterTable alterTable = new ClickHouseAlterTable(); + String result = alterTable.createAlterTableSyntax( + "test_db", "test_table", + null, + ClickHouseAlterTable.ALTER_TABLE_OPERATION.ADD); + assertEquals("", result, "Null column map should return empty string"); + } - // Delete Column - String alterTableDeleteColumnQuery = cat.createAlterTableSyntax("employees", - this.getExpectedColumnToDataTypesMap(), ClickHouseAlterTable.ALTER_TABLE_OPERATION.REMOVE); + @Test + @DisplayName("Single column ADD produces valid ALTER TABLE SQL") + public void testSingleColumnAdd() { + ClickHouseAlterTable alterTable = new ClickHouseAlterTable(); + Map cols = new LinkedHashMap<>(); + cols.put("new_col", "String"); + String result = alterTable.createAlterTableSyntax( + "test_table", + cols, + ClickHouseAlterTable.ALTER_TABLE_OPERATION.ADD); + assertTrue(result.contains("ALTER TABLE"), "Should contain ALTER TABLE"); + assertTrue(result.contains("`new_col`"), "Should backtick-escape column name"); + assertTrue(result.contains("add column"), "Should contain add column"); + } - String expectedDeleteColumnQuery = "ALTER TABLE employees delete column `amount` Float64,delete column `occupation` String,delete column `quantity` Int32,delete column `blob_storage_scale` Decimal,delete column `json_output` JSON,delete column `max_amount` Float64,delete column `amount_1` Float32,delete column `customerName` String,delete column `blob_storage` String,delete column `employed` Bool"; - Assert.assertTrue(alterTableDeleteColumnQuery.equalsIgnoreCase(expectedDeleteColumnQuery)); + @Test + @DisplayName("Database-prefixed ALTER TABLE includes backtick-escaped database.table") + public void testDatabasePrefixedAlterTable() { + ClickHouseAlterTable alterTable = new ClickHouseAlterTable(); + Map cols = new LinkedHashMap<>(); + cols.put("col1", "Int32"); + String result = alterTable.createAlterTableSyntax( + "my_db", "my_table", + cols, + ClickHouseAlterTable.ALTER_TABLE_OPERATION.ADD); + assertTrue(result.contains("`my_db`.`my_table`"), + "Should contain backtick-escaped database.table: " + result); + } + + @Test + @DisplayName("Multiple columns produce comma-separated ADD COLUMN clauses") + public void testMultipleColumnsAdd() { + ClickHouseAlterTable alterTable = new ClickHouseAlterTable(); + Map cols = new LinkedHashMap<>(); + cols.put("col_a", "String"); + cols.put("col_b", "Int64"); + String result = alterTable.createAlterTableSyntax( + "test_table", + cols, + ClickHouseAlterTable.ALTER_TABLE_OPERATION.ADD); + assertTrue(result.contains("`col_a`"), "Should contain col_a"); + assertTrue(result.contains("`col_b`"), "Should contain col_b"); + // Should not end with comma + assertFalse(result.trim().endsWith(","), "Should not end with trailing comma"); } } diff --git a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseAutoCreateTableTest.java b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseAutoCreateTableTest.java index 214c7ca05..ac1034698 100644 --- a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseAutoCreateTableTest.java +++ b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseAutoCreateTableTest.java @@ -34,7 +34,7 @@ public void testCreateTableSyntax() { String query = act.createTableSyntax(primaryKeys, "auto_create_table", "employees", createFields(), this.columnToDataTypesMap, false, false, null,new ClickHouseSinkConnectorConfig(new HashMap<>())); System.out.println("QUERY" + query); - Assert.assertTrue(query.equalsIgnoreCase("CREATE TABLE employees.`auto_create_table`(`customerName` String NOT NULL,`occupation` String NOT NULL,`quantity` Int32 NOT NULL,`amount_1` Float32 NOT NULL,`amount` Float64 NOT NULL,`employed` Bool NOT NULL,`blob_storage` String NOT NULL,`blob_storage_scale` Decimal NOT NULL,`json_output` JSON,`max_amount` Float64 NOT NULL,`_sign` Int8,`_version` UInt64) ENGINE = ReplacingMergeTree(_version) PRIMARY KEY(customerName) ORDER BY(customerName)")); + Assert.assertTrue(query.equalsIgnoreCase("CREATE TABLE IF NOT EXISTS `employees`.`auto_create_table`(`customerName` String NOT NULL,`occupation` String NOT NULL,`quantity` Int32 NOT NULL,`amount_1` Float32 NOT NULL,`amount` Float64 NOT NULL,`employed` Bool NOT NULL,`blob_storage` String NOT NULL,`blob_storage_scale` Decimal NOT NULL,`json_output` JSON,`max_amount` Float64 NOT NULL,`_sign` Int8,`_version` UInt64) ENGINE = ReplacingMergeTree(_version) PRIMARY KEY(`customerName`) ORDER BY(`customerName`)")); //Assert.assertTrue(query.equalsIgnoreCase("CREATE TABLE auto_create_table(`customerName` String NOT NULL,`occupation` String NOT NULL,`quantity` Int32 NOT NULL,`amount_1` Float32 NOT NULL,`amount` Float64 NOT NULL,`employed` Bool NOT NULL,`blob_storage` String NOT NULL,`blob_storage_scale` Decimal NOT NULL,`json_output` JSON,`max_amount` Float64 NOT NULL,`_sign` Int8,`_version` UInt64) ENGINE = ReplacingMergeTree(_version) PRIMARY KEY(customerName) ORDER BY (customerName)")); } @@ -46,7 +46,7 @@ public void testCreateTableEmptyPrimaryKey() { String query = act.createTableSyntax(null, "auto_create_table", "employees", createFields(), this.columnToDataTypesMap, false, false, null,new ClickHouseSinkConnectorConfig(new HashMap<>())); - String expectedQuery = "CREATE TABLE employees.`auto_create_table`(`customerName` String NOT NULL,`occupation` String NOT NULL,`quantity` Int32 NOT NULL,`amount_1` Float32 NOT NULL,`amount` Float64 NOT NULL,`employed` Bool NOT NULL,`blob_storage` String NOT NULL,`blob_storage_scale` Decimal NOT NULL,`json_output` JSON,`max_amount` Float64 NOT NULL,`_sign` Int8,`_version` UInt64) ENGINE = ReplacingMergeTree(_version) ORDER BY tuple()"; + String expectedQuery = "CREATE TABLE IF NOT EXISTS `employees`.`auto_create_table`(`customerName` String NOT NULL,`occupation` String NOT NULL,`quantity` Int32 NOT NULL,`amount_1` Float32 NOT NULL,`amount` Float64 NOT NULL,`employed` Bool NOT NULL,`blob_storage` String NOT NULL,`blob_storage_scale` Decimal NOT NULL,`json_output` JSON,`max_amount` Float64 NOT NULL,`_sign` Int8,`_version` UInt64) ENGINE = ReplacingMergeTree(_version) ORDER BY tuple()"; Assert.assertTrue(query.equalsIgnoreCase(expectedQuery)); } @Test @@ -60,7 +60,7 @@ public void testCreateTableMultiplePrimaryKeys() { String query = act.createTableSyntax(primaryKeys, "auto_create_table", "customers", createFields(), this.columnToDataTypesMap, false, false, null,new ClickHouseSinkConnectorConfig(new HashMap<>())); - String expectedQuery = "CREATE TABLE customers.`auto_create_table`(`customerName` String NOT NULL,`occupation` String NOT NULL,`quantity` Int32 NOT NULL,`amount_1` Float32 NOT NULL,`amount` Float64 NOT NULL,`employed` Bool NOT NULL,`blob_storage` String NOT NULL,`blob_storage_scale` Decimal NOT NULL,`json_output` JSON,`max_amount` Float64 NOT NULL,`_sign` Int8,`_version` UInt64) ENGINE = ReplacingMergeTree(_version) ORDER BY tuple()"; + String expectedQuery = "CREATE TABLE IF NOT EXISTS `customers`.`auto_create_table`(`customerName` String NOT NULL,`occupation` String NOT NULL,`quantity` Int32 NOT NULL,`amount_1` Float32 NOT NULL,`amount` Float64 NOT NULL,`employed` Bool NOT NULL,`blob_storage` String NOT NULL,`blob_storage_scale` Decimal NOT NULL,`json_output` JSON,`max_amount` Float64 NOT NULL,`_sign` Int8,`_version` UInt64) ENGINE = ReplacingMergeTree(_version) ORDER BY tuple()"; Assert.assertTrue(query.equalsIgnoreCase(expectedQuery)); System.out.println(query); }