Add a ClickHouse connector with batched writes

[GabrielBBaldez]

What

Adds geaflow-dsl-connector-clickhouse, a ClickHouse connector with a batched sink and a partitioned source, wired in through the connector SPI and modelled on geaflow-dsl-connector-jdbc.

Fixes #798

Sink — batched writes

Rows are buffered and flushed in bulk via PreparedStatement.addBatch()/executeBatch() once geaflow.dsl.clickhouse.write.batch.size (default 1000) is reached, and again when the window finishes. ClickHouse is columnar and optimized for bulk ingestion, so this is far faster than the row-by-row inserts you get when connecting through the generic JDBC connector.

Source — parallel partitioned reads

The read is split over a numeric column into partition.num ranges (the same partitioning model as the JDBC connector) so the partitions can be read in parallel.

Tested

Testcontainers integration test against a real ClickHouse 24.3 instance:

• batched write + row-count verification, partitioned read round-trip, partition splitting, and SPI discovery (type='clickhouse' resolving to the connector);
• a write benchmark — batched vs. row-by-row → roughly 30x higher throughput (logged by the test).

checkstyle and RAT pass. Chinese and English docs added under docs/docs-cn / docs/docs-en and wired into the connector index.rst.

[kitalkuyo-gita]

@GabrielBBaldez Hello! I’ve been extremely busy lately and have just seen your PR. Thank you very much for your valuable contribution to the community. We take a rigorous and serious approach with every new contributor; I have carefully reviewed your code and identified some performance issues, so I hope you can take a moment to look at them. Best regards.

[kitalkuyo-gita]

Incorrect partition stride calculation — potential for overlapping partitions or missed reads

Code snippet: ClickHouseTableSource.listPartitions()

Issue

The current code—long stride = upperBound / partitionNum - lowerBound / partitionNum;—splits the division into two separate integer divisions before subtracting, resulting in rounding errors; semantically, it should be (upperBound - lowerBound) / partitionNum.

Risk

Incorrect partition boundary calculations can lead to rows being read multiple times or not at all (due to erroneous range splitting).

Solution (fix and handle small/zero ranges):

Use long range = upperBound - lowerBound; stride = Math.max(1, range / partitionNum);
Ensure partitionNum does not exceed range (if range < partitionNum, cap it at range or 1), and ensure all rows are covered in edge cases (with the last partition absorbing the remainder). #### Suggested code (replacing the original stride/loop section):

long range = upperBound - lowerBound;
if (range <= 0) {
throw new GeaFlowDSLException("invalid partition range: lowerbound=%d upperbound=%d", lowerBound, upperBound);
}
if (partitionNum > range) {
partitionNum = range; // at most one id per partition
}
long baseStride = range / partitionNum;
long remainder = range % partitionNum;
long current = lowerBound;
List<Partition> partitions = new ArrayList<>();
for (long i = 0; i < partitionNum; ++i) {
long thisStride = baseStride + (i < remainder ? 1 : 0);
long next = current + thisStride;
String lBound = (i != 0) ? String.format("%s >= %d", partitionColumn, current) : null;
String uBound = (i != partitionNum - 1) ? String.format("%s < %d", partitionColumn, next) : null;
String whereClause;
if (uBound == null) {
whereClause = lBound;
} else if (lBound == null) {
whereClause = String.format("%s OR %s IS NULL", uBound, partitionColumn);
} else {
whereClause = String.format("%s AND %s", lBound, uBound);
}
partitions.add(new ClickHousePartition(tableName, "WHERE " + whereClause));
current = next;
}

Explanation: Here, the remainder is distributed among the first few partitions, ensuring the entire range is fully covered without overlap.

[kitalkuyo-gita]

SQL Construction and Identifier/Parameter Security — Risks of SQL Injection and Identifier Injection

Code references: ClickHouseUtils.buildInsertSql(), ClickHouseUtils.selectRowsFromTable(), ClickHouseTableSource.listPartitions() (involving the construction of whereClause and the passing of table/column names).

Issue

Table and column names are directly concatenated into the SQL string without validation or quoting. If a user injects a malicious string (or one containing spaces or special characters) via the configuration, it can lead to syntax errors or injection vulnerabilities.
selectRowsFromTable uses Statement combined with string concatenation for clauses such as LIMIT, OFFSET, and ORDER BY.

Risk

If the configuration source—even if generally trusted—unexpectedly contains special characters, it could cause query failures or security issues. The general rule is that all external inputs, including configurations, must be validated or escaped.

Solution (Two Parts)

A) Validate table and column names against an allowlist and enclose them in identifier quotes (ClickHouse supports backticks or double quotes; choose the style consistent with the project) to reject invalid identifiers. B) Switch to using PreparedStatement for LIMIT and OFFSET values ​​(passing parameterizable numbers as parameters) and avoid concatenating variable strings into the whereClause whenever possible. If the whereClause is constructed internally by the framework (as is currently the case), ensure that only numeric constants are inserted, or use placeholders and set the corresponding parameters. Suggested universal tool method (put in ClickHouseUtils):

private static final Pattern IDENTIFIER = Pattern.compile("^[A-Za-z_][A-Za-z0-9_]*$");

public static String quoteIdentifier(String id) { 
if (id == null || !IDENTIFIER.matcher(id).matches()) { 
throw new GeaFlowDSLException("invalid identifier: " + id); 
} 
return "`" + id + "`"; // ClickHouse accepts backticks
}
Modify buildInsertSql to use quoteIdentifier:
Java
public static String buildInsertSql(String tableName, List<TableField> fields) { 
String qTable = quoteIdentifier(tableName); 
String columns = fields.stream() 
.map(f -> quoteIdentifier(f.getName())) 
.collect(Collectors.joining(", ")); 
String placeholders = fields.stream().map(f -> "?").collect(Collectors.joining(", ")); 
return String.format("INSERT INTO %s (%s) VALUES (%s)", qTable, columns, placeholders);
}

Modify selectRowsFromTable to use PreparedStatement (and use quoteIdentifier to verify the table name/column name):

public static List<Row> selectRowsFromTable(Connection conn, String tableName, 
String whereClause, int columnNum, 
long startOffset, long windowSize, String orderByColumnName) 
throws SQLException { 
if (windowSize == Windows.SIZE_OF_ALL_WINDOW) { 
windowSize = Integer.MAX_VALUE; 
} else if (windowSize <= 0) { 
throw new GeaFlowDSLException("wrong windowSize"); 
} 

String qTable = quoteIdentifier(tableName); 
String qOrder = quoteIdentifier(orderByColumnName); 
// whereClause is either empty or starts with "WHERE ...". If it was constructed here (numbers), 
// it's safe. Otherwise validate/deny. 
String sql = String.format("SELECT * FROM %s %s ORDER BY %s LIMIT ? OFFSET ?", qTable, whereClause == null ? "" : whereClause, qOrder); 
try (PreparedStatement ps = conn.prepareStatement(sql)) { 
ps.setLong(1, windowSize);
ps.setLong(2, startOffset);
try (ResultSet rs = ps.executeQuery()) {
List<Row> rowList = new ArrayList<>();
while (rs.next()) {
Object[] values ​​= new Object[columnNum];
for (int i = 1; i <= columnNum; i++) {
values[i - 1] = rs.getObject(i);
}
rowList.add(ObjectRow.create(values));
}
return rowList;
}
}
}

Note: PreparedStatement cannot be used to substitute table or column names, so quoteIdentifier validation is crucial.

[kitalkuyo-gita]

Code reference: ClickHouseTableSink.flush() — executeBatch may throw an SQLException; bufferedRows is not rolled back, and statement.clearBatch() is not executed when an exception occurs.

• ClickHouseTableSink.close(): If close throws an exception, it masks exceptions thrown upstream (e.g., errors during finish), preventing tests or users from seeing the original error.
• ClickHouseTableSource.fetch(): Uses Statement/Connection; specifically, selectRowsFromTable calls statement.executeQuery(selectQuery) and closes the ResultSet in a utility method, whereas PreparedStatement usage is best handled via a local try-with-resources block.

Risks: Resource leaks (open connections/statements), unclear error logs, loss of buffered row count state following batch failures, and difficulties with program retries or recovery. Suggested code modifications (flush & close):

private void flush() throws SQLException {
if (bufferedRows == 0) {
return;
}
try {
statement.executeBatch();
// If ClickHouse driver requires commit, optionally:
// if (!connection.getAutoCommit()) connection.commit();
LOGGER.info("flushed {} rows to clickhouse table {}", bufferedRows, tableName);
} finally {
try {
statement.clearBatch();
} catch (SQLException e) {
LOGGER.warn("failed to clear batch", e);
}
bufferedRows = 0;
}
}

@Override
public void close() {
SQLException closeEx = null;
try {
if (this.statement != null) {
try {
this.statement.close();
} catch (SQLException e) {
LOGGER.warn("failed to close statement", e);
closeEx = e;
} finally {
this.statement = null;
}
}
if (this.connection != null) {
try {
this.connection.close();
} catch (SQLException e) {
LOGGER.warn("failed to close connection", e);
if (closeEx == null) closeEx = e;
} finally {
this.connection = null;
}
}
LOGGER.info("close clickhouse sink");
} finally {
if (closeEx != null) {
// Do not swallow—wrap in a runtime exception to surface it to the caller, or log it; ensure previous exceptions are not lost.
throw new GeaFlowDSLException("failed to close clickhouse sink", closeEx);
}
}
}

Explanation: The example above attempts to capture exceptions encountered while closing each resource and ultimately throws them as a single exception (or, depending on project conventions, logs them as WARN and suppresses them).