Skip to content

[split 16/22] db ops: QueryFormatter destination-column awareness, ALTER/CREATE quoting fixes #1371

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
minguyen9988 wants to merge 1 commit into
base: omniwatcher/split-15-sink-config
Choose a base branch
from
Open

[split 16/22] db ops: QueryFormatter destination-column awareness, ALTER/CREATE quoting fixes #1371

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 23 additions & 3 deletions ...nector/src/main/java/com/altinity/clickhouse/sink/connector/db/ClickHouseDbConstants.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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, "
Expand Down
47 changes: 26 additions & 21 deletions ...onnector/src/main/java/com/altinity/clickhouse/sink/connector/db/DbKafkaOffsetWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
*/
Expand Down Expand Up @@ -61,6 +66,7 @@ public DbKafkaOffsetWriter(
super(hostName, port, database, userName, password, config,
connection);

this.tableName = tableName;
createOffsetTable();
this.columnNamesToDataTypesMap =
new DBMetadata(config).getColumnsDataTypesForTable(
Expand All @@ -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);
}
}

Expand Down Expand Up @@ -148,22 +152,23 @@ public void insertTopicOffsetMetadata(
public Map<TopicPartition, Long> getStoredOffsets() throws SQLException {
Map<TopicPartition, Long> 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;
Expand Down
113 changes: 105 additions & 8 deletions sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/QueryFormatter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,10 +79,35 @@ public MutablePair<String, Map<String, Integer>> 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<String, Map<String, Integer>> getInsertQueryUsingInputFunction(
String tableName, List<Field> fields,
Map<String, String> columnNameToDataTypeMap,
boolean includeKafkaMetaData,
boolean includeRawData,
String rawDataColumn, String dbName,
java.util.Collection<String> 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;
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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();
}

/**
Expand Down Expand Up @@ -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 + "')";
}
Expand Down Expand Up @@ -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<Field> fields, Map<String, String> columnNameToDataTypeMap,
boolean includeKafkaMetaData, boolean includeRawData, String rawDataColumn, String dbName) {
boolean includeKafkaMetaData, boolean includeRawData, String rawDataColumn, String dbName,
java.util.Collection<String> connectorManagedColumns) {

if (fields == null) {
log.error("getInsertQueryUsingInputFunction, fields empty");
Expand All @@ -283,6 +341,26 @@ private ColumnData createColumns(String tableName, List<Field> fields, Map<Strin
StringBuilder colNamesDelimited = new StringBuilder();
StringBuilder colNamesToDataTypes = new StringBuilder();

// Column names present in THIS record group's source schema
// (lower-cased for case-insensitive membership checks).
java.util.Set<String> 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<String> 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<String, String> entry : columnNameToDataTypeMap.entrySet()) {
String sourceColumnName = entry.getKey();
Expand All @@ -308,6 +386,25 @@ private ColumnData createColumns(String tableName, List<Field> fields, Map<Strin
colNamesToDataTypes.append(sourceColumnNameWithBackTicks).append(" ").append(dataType).append(",");
colNameToIndexMap.put(sourceColumnName, index++);
}
} else if (!sourceFieldNamesLower.contains(
sourceColumnName.toLowerCase(java.util.Locale.ROOT))
&& !isConnectorManagedColumn(sourceColumnName)
&& !managedLower.contains(
sourceColumnName.toLowerCase(java.util.Locale.ROOT))) {
// Destination-only DATA column: the source event does not
// carry a value for it (e.g. a column ADDed by a later DDL,
// with pre-ALTER records still in flight). It must be
// OMITTED from the insert so ClickHouse fills its DEFAULT.
// Including it forced a NULL bind, which fails outright on
// non-nullable columns — the whole batch then errors and
// replays forever, stalling replication (observed: the
// price_usd ADD COLUMN scenario, 35 of 200 rows never
// landing). Connector-managed columns (_version, _sign,
// is_deleted, temporal tracking) are always included: the
// field mapper computes their values itself.
log.debug("Omitting destination-only column {} from insert into {}.{} "
+ "(not present in source event; ClickHouse DEFAULT applies)",
sourceColumnName, dbName, tableName);
} else {
colNamesDelimited.append(sourceColumnNameWithBackTicks).append(",");
colNamesToDataTypes.append(sourceColumnNameWithBackTicks).append(" ").append(dataType).append(",");
Expand Down
29 changes: 27 additions & 2 deletions .../main/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseAlterTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,10 +68,35 @@ public enum ALTER_TABLE_OPERATION {
public String createAlterTableSyntax(String tableName,
Map<String, String> 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<String, String> 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<String, String> entry
: colNameToDataTypesMap.entrySet()) {
Expand Down
8 changes: 4 additions & 4 deletions .../java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseAutoCreateTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -124,7 +124,7 @@ public String createTableSyntax(ArrayList<String> 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}` ");
Expand Down Expand Up @@ -249,12 +249,12 @@ public String createTableSyntax(ArrayList<String> 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("`");
Expand Down Expand Up @@ -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));
Expand Down
4 changes: 3 additions & 1 deletion ...n/java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseCreateDatabase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down
4 changes: 2 additions & 2 deletions ...a/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseTableOperationsBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -89,7 +89,7 @@ public ClickHouseTableOperationsBase() {
*/
public Map<String, String> getColumnNameToCHDataTypeMapping(Field[] fields, ClickHouseSinkConnectorConfig config) {
ClickHouseDataTypeMapper mapper = new ClickHouseDataTypeMapper();
Map<String, String> columnToDataTypesMap = new HashMap<>();
Map<String, String> columnToDataTypesMap = new LinkedHashMap<>();

for (Field f : fields) {
String colName = f.name();
Expand Down
Loading
Loading