This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 0580c092b0 [Fix][Connector-V2] Preserve offset timestamps in
ClickHouse sink statements (#12285)
0580c092b0 is described below
commit 0580c092b0ac2a2bfae94cb17f98a41390421077
Author: Goutam Adwant <[email protected]>
AuthorDate: Sun Sep 13 06:19:28 2026 +0000
[Fix][Connector-V2] Preserve offset timestamps in ClickHouse sink
statements (#12285)
---
docs/en/connectors/sink/Clickhouse.md | 8 +
docs/zh/connectors/sink/Clickhouse.md | 8 +
.../executor/FieldNamedPreparedStatement.java | 21 +-
.../JdbcBatchStatementExecutorBuilder.java | 19 +-
.../sink/client/executor/JdbcRowConverter.java | 31 +++
.../sink/inject/OffsetDateTimeInjectFunction.java | 62 +++++
.../sink/client/executor/JdbcRowConverterTest.java | 204 ++++++++++++++++
.../clickhouse/ClickhouseOffsetTimestampIT.java | 269 +++++++++++++++++++++
8 files changed, 609 insertions(+), 13 deletions(-)
diff --git a/docs/en/connectors/sink/Clickhouse.md
b/docs/en/connectors/sink/Clickhouse.md
index 440a16d0b8..584cb4eddb 100644
--- a/docs/en/connectors/sink/Clickhouse.md
+++ b/docs/en/connectors/sink/Clickhouse.md
@@ -48,6 +48,14 @@ They can be downloaded via install-plugin.sh or from the
Maven central repositor
| ARRAY | Array
|
| MAP | Map
|
+### Timezone-aware timestamps
+
+Scalar `TIMESTAMP_TZ` values can be written to existing `DateTime` and
`DateTime64` columns, including nullable columns. The sink preserves the
instant represented by the input offset. `DateTime` stores whole seconds;
`DateTime64` stores fractional seconds up to the target column's precision.
ClickHouse uses a column-level timezone and does not retain the original offset
of each row.
+
+The sink uses explicit UTC conversions for these fields, including update and
delete conditions. Inserts containing these fields use the driver's SQL-based
batching instead of its binary input path. Inserts without these fields retain
the existing path.
+
+Create the target table before running the job. Automatic table creation for
`TIMESTAMP_TZ` and timezone-aware timestamps inside arrays or maps are not
supported by this mapping.
+
## Sink Options
| Name | Type | Required | Default |
Description
|
diff --git a/docs/zh/connectors/sink/Clickhouse.md
b/docs/zh/connectors/sink/Clickhouse.md
index fb97277384..80a3a9c1bc 100644
--- a/docs/zh/connectors/sink/Clickhouse.md
+++ b/docs/zh/connectors/sink/Clickhouse.md
@@ -47,6 +47,14 @@ import ChangeLog from '../changelog/connector-clickhouse.md';
| ARRAY | Array
|
| MAP | Map
|
+### 带时区的时间戳
+
+标量 `TIMESTAMP_TZ` 值可以写入已存在的 `DateTime` 和 `DateTime64` 列，包括可空列。Sink
保留输入偏移量所表示的时间点。`DateTime` 存储整秒，`DateTime64` 按目标列的精度存储小数秒。ClickHouse
使用列级别的时区，不保留每行原始的时区偏移量。
+
+Sink 对这些字段使用显式 UTC 转换，包括更新和删除条件。包含这些字段的插入使用驱动的 SQL
批处理路径，而不是二进制输入路径。不包含这些字段的插入保持原有路径不变。
+
+运行作业前请先创建目标表。此映射不支持为 `TIMESTAMP_TZ` 自动建表，也不支持数组或 Map 中的带时区时间戳。
+
## Sink 选项
| 名称 | 类型 | 是否必须 | 默认值 |
描述
|
diff --git
a/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/FieldNamedPreparedStatement.java
b/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/FieldNamedPreparedStatement.java
index 5fa82e8c59..da543ad5db 100644
---
a/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/FieldNamedPreparedStatement.java
+++
b/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/FieldNamedPreparedStatement.java
@@ -45,6 +45,7 @@ import java.util.Calendar;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.function.Function;
import static
org.apache.seatunnel.shade.com.google.common.base.Preconditions.checkArgument;
import static
org.apache.seatunnel.shade.com.google.common.base.Preconditions.checkNotNull;
@@ -618,6 +619,15 @@ public class FieldNamedPreparedStatement implements
PreparedStatement {
public static FieldNamedPreparedStatement prepareStatement(
Connection connection, String sql, String[] fieldNames) throws
SQLException {
+ return prepareStatement(connection, sql, fieldNames, field -> "?");
+ }
+
+ static FieldNamedPreparedStatement prepareStatement(
+ Connection connection,
+ String sql,
+ String[] fieldNames,
+ Function<String, String> parameterExpression)
+ throws SQLException {
checkNotNull(connection, "connection must not be null.");
checkNotNull(sql, "sql must not be null.");
checkNotNull(fieldNames, "fieldNames must not be null.");
@@ -632,7 +642,7 @@ public class FieldNamedPreparedStatement implements
PreparedStatement {
}
} else {
HashMap<String, List<Integer>> parameterMap = new HashMap<>();
- parsedSQL = parseNamedStatement(sql, parameterMap);
+ parsedSQL = parseNamedStatement(sql, parameterMap,
parameterExpression);
// currently, the statements must contain all the field parameters
checkArgument(parameterMap.size() >= fieldNames.length);
for (int i = 0; i < fieldNames.length; i++) {
@@ -648,6 +658,13 @@ public class FieldNamedPreparedStatement implements
PreparedStatement {
}
public static String parseNamedStatement(String sql, Map<String,
List<Integer>> paramMap) {
+ return parseNamedStatement(sql, paramMap, field -> "?");
+ }
+
+ private static String parseNamedStatement(
+ String sql,
+ Map<String, List<Integer>> paramMap,
+ Function<String, String> parameterExpression) {
StringBuilder parsedSql = new StringBuilder();
int fieldIndex = 1; // SQL statement parameter index starts from 1
int length = sql.length();
@@ -667,7 +684,7 @@ public class FieldNamedPreparedStatement implements
PreparedStatement {
paramMap.computeIfAbsent(parameterName, n -> new
ArrayList<>()).add(fieldIndex);
fieldIndex++;
i = j - 1;
- parsedSql.append('?');
+ parsedSql.append(parameterExpression.apply(parameterName));
} else {
parsedSql.append(c);
}
diff --git
a/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/JdbcBatchStatementExecutorBuilder.java
b/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/JdbcBatchStatementExecutorBuilder.java
index 3a31915bba..5ffc72bddd 100644
---
a/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/JdbcBatchStatementExecutorBuilder.java
+++
b/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/JdbcBatchStatementExecutorBuilder.java
@@ -183,12 +183,12 @@ public class JdbcBatchStatementExecutorBuilder {
JdbcRowConverter rowConverter) {
return new InsertOrUpdateBatchStatementExecutor(
connection ->
- FieldNamedPreparedStatement.prepareStatement(
+ rowConverter.prepareStatement(
connection,
SqlUtils.getInsertIntoStatement(table,
rowType.getFieldNames()),
rowType.getFieldNames()),
connection ->
- FieldNamedPreparedStatement.prepareStatement(
+ rowConverter.prepareStatement(
connection,
SqlUtils.getAlterTableUpdateStatement(
table, rowType.getFieldNames(),
pkNames),
@@ -205,17 +205,17 @@ public class JdbcBatchStatementExecutorBuilder {
JdbcRowConverter valueConverter) {
return new InsertOrUpdateBatchStatementExecutor(
connection ->
- FieldNamedPreparedStatement.prepareStatement(
+ keyConverter.prepareStatement(
connection,
SqlUtils.getRowExistsStatement(table, pkNames),
pkNames),
connection ->
- FieldNamedPreparedStatement.prepareStatement(
+ valueConverter.prepareStatement(
connection,
SqlUtils.getInsertIntoStatement(table,
rowType.getFieldNames()),
rowType.getFieldNames()),
connection ->
- FieldNamedPreparedStatement.prepareStatement(
+ valueConverter.prepareStatement(
connection,
SqlUtils.getAlterTableUpdateStatement(
table, rowType.getFieldNames(),
pkNames),
@@ -230,7 +230,7 @@ public class JdbcBatchStatementExecutorBuilder {
String insertSQL = SqlUtils.getInsertIntoStatement(table,
rowType.getFieldNames());
return new SimpleBatchStatementExecutor(
connection ->
- FieldNamedPreparedStatement.prepareStatement(
+ rowConverter.prepareStatement(
connection, insertSQL,
rowType.getFieldNames()),
rowConverter);
}
@@ -244,9 +244,7 @@ public class JdbcBatchStatementExecutorBuilder {
SqlUtils.getDeleteStatement(
table, primaryKeys,
enableExperimentalLightweightDelete);
return new SimpleBatchStatementExecutor(
- connection ->
- FieldNamedPreparedStatement.prepareStatement(
- connection, deleteSQL, primaryKeys),
+ connection -> rowConverter.prepareStatement(connection,
deleteSQL, primaryKeys),
rowConverter);
}
@@ -255,8 +253,7 @@ public class JdbcBatchStatementExecutorBuilder {
String alterTableDeleteSQL =
SqlUtils.getAlterTableDeleteStatement(table, primaryKeys);
return new SimpleBatchStatementExecutor(
connection ->
- FieldNamedPreparedStatement.prepareStatement(
- connection, alterTableDeleteSQL, primaryKeys),
+ rowConverter.prepareStatement(connection,
alterTableDeleteSQL, primaryKeys),
rowConverter);
}
diff --git
a/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/JdbcRowConverter.java
b/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/JdbcRowConverter.java
index d091a9e881..a579a88bff 100644
---
a/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/JdbcRowConverter.java
+++
b/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/JdbcRowConverter.java
@@ -19,6 +19,7 @@ package
org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.client.executo
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.api.table.type.SqlType;
import
org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.inject.ArrayInjectFunction;
import
org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.inject.BigDecimalInjectFunction;
import
org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.inject.ClickhouseFieldInjectFunction;
@@ -29,11 +30,13 @@ import
org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.inject.FloatInj
import
org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.inject.IntInjectFunction;
import
org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.inject.LongInjectFunction;
import
org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.inject.MapInjectFunction;
+import
org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.inject.OffsetDateTimeInjectFunction;
import
org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.inject.StringInjectFunction;
import lombok.NonNull;
import java.io.Serializable;
+import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
@@ -60,9 +63,37 @@ public class JdbcRowConverter implements Serializable {
this.projectionFields = projectionFields;
this.fieldInjectFunctionMap =
createFieldInjectFunctionMap(projectionFields,
clickhouseTableSchema);
+ for (String field : projectionFields) {
+ if (rowType.getFieldType(rowType.indexOf(field)).getSqlType() ==
SqlType.TIMESTAMP_TZ
+ && fieldInjectFunctionMap.get(field) instanceof
DateTimeInjectFunction) {
+ fieldInjectFunctionMap.put(
+ field,
+ new OffsetDateTimeInjectFunction(
+
unwrapCommonPrefix(clickhouseTableSchema.get(field))));
+ }
+ }
this.fieldGetterMap = createFieldGetterMap(projectionFields, rowType);
}
+ PreparedStatement prepareStatement(Connection connection, String sql,
String[] fields)
+ throws SQLException {
+ if (sql.contains("?")
+ && fieldInjectFunctionMap.values().stream()
+
.anyMatch(OffsetDateTimeInjectFunction.class::isInstance)) {
+ throw new IllegalArgumentException("TIMESTAMP_TZ requires named
SQL parameters");
+ }
+ return FieldNamedPreparedStatement.prepareStatement(
+ connection,
+ sql,
+ fields,
+ field -> {
+ ClickhouseFieldInjectFunction function =
fieldInjectFunctionMap.get(field);
+ return function instanceof OffsetDateTimeInjectFunction
+ ? ((OffsetDateTimeInjectFunction)
function).getParameterExpression()
+ : "?";
+ });
+ }
+
public PreparedStatement toExternal(SeaTunnelRow row, PreparedStatement
statement)
throws SQLException {
for (int i = 0; i < projectionFields.length; i++) {
diff --git
a/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/inject/OffsetDateTimeInjectFunction.java
b/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/inject/OffsetDateTimeInjectFunction.java
new file mode 100644
index 0000000000..c72d6deee2
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/inject/OffsetDateTimeInjectFunction.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.inject;
+
+import com.clickhouse.client.ClickHouseColumn;
+
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeFormatterBuilder;
+import java.time.temporal.ChronoField;
+
+/** Binds UTC text together with an explicitly UTC SQL conversion to preserve
the instant. */
+public class OffsetDateTimeInjectFunction extends DateTimeInjectFunction {
+ private static final DateTimeFormatter SECONDS =
+ DateTimeFormatter.ofPattern("uuuu-MM-dd HH:mm:ss");
+ private static final DateTimeFormatter FRACTIONS =
+ new DateTimeFormatterBuilder()
+ .append(SECONDS)
+ .appendFraction(ChronoField.NANO_OF_SECOND, 0, 9, true)
+ .toFormatter();
+ private final boolean dateTime64;
+ private final String parameterExpression;
+
+ public OffsetDateTimeInjectFunction(String targetType) {
+ dateTime64 = targetType.startsWith("DateTime64");
+ parameterExpression =
+ dateTime64
+ ? "toDateTime64(?, "
+ + ClickHouseColumn.of("timestamp",
targetType).getScale()
+ + ", 'UTC')"
+ : "toDateTime(?, 'UTC')";
+ }
+
+ public String getParameterExpression() {
+ return parameterExpression;
+ }
+
+ @Override
+ public void injectFields(PreparedStatement statement, int index, Object
value)
+ throws SQLException {
+ OffsetDateTime utc = ((OffsetDateTime)
value).withOffsetSameInstant(ZoneOffset.UTC);
+ statement.setString(index, (dateTime64 ? FRACTIONS :
SECONDS).format(utc));
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-clickhouse/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/JdbcRowConverterTest.java
b/seatunnel-connectors-v2/connector-clickhouse/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/JdbcRowConverterTest.java
new file mode 100644
index 0000000000..a34d11fc83
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-clickhouse/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/JdbcRowConverterTest.java
@@ -0,0 +1,204 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package
org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.client.executor;
+
+import org.apache.seatunnel.api.table.type.LocalTimeType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+class JdbcRowConverterTest {
+
+ @ParameterizedTest
+ @ValueSource(
+ strings = {
+ "DateTime",
+ "DateTime('Asia/Shanghai')",
+ "DateTime64(0)",
+ "DateTime64(3)",
+ "DateTime64(9, 'UTC')",
+ "Nullable(DateTime64(6, 'America/New_York'))"
+ })
+ void bindsOffsetDateTimeWithoutDiscardingOffsetOrPrecision(String
targetType) throws Exception {
+ JdbcRowConverter converter = converter(targetType);
+ for (String literal :
+ new String[] {
+ "2026-09-12T10:00:00.123456789Z",
+ "2026-09-12T10:00:00.123456789+05:30",
+ "2026-09-12T10:00:00.123456789-07:00"
+ }) {
+ PreparedStatement statement = mock(PreparedStatement.class);
+ OffsetDateTime value = OffsetDateTime.parse(literal);
+ converter.toExternal(new SeaTunnelRow(new Object[] {value}),
statement);
+ String expected =
+ value.withOffsetSameInstant(ZoneOffset.UTC)
+ .format(DateTimeFormatter.ofPattern("uuuu-MM-dd
HH:mm:ss"));
+ if (targetType.contains("DateTime64")) {
+ expected += ".123456789";
+ }
+ verify(statement).setString(1, expected);
+ verifyNoMoreInteractions(statement);
+ }
+ }
+
+ @Test
+ void preservesExistingTimestampBindingsAndNulls() throws Exception {
+ JdbcRowConverter converter =
+ new JdbcRowConverter(
+ new SeaTunnelRowType(
+ new String[] {"event_time"},
+ new SeaTunnelDataType[]
{LocalTimeType.LOCAL_DATE_TIME_TYPE}),
+ Collections.singletonMap("event_time",
"Nullable(DateTime64(9, 'UTC'))"),
+ new String[] {"event_time"});
+ LocalDateTime local =
LocalDateTime.parse("2026-09-12T10:00:00.123456789");
+ Timestamp timestamp = Timestamp.valueOf(local);
+ for (Object value :
+ new Object[] {local, timestamp, "2026-09-12
10:00:00.123456789", null}) {
+ PreparedStatement statement = mock(PreparedStatement.class);
+ converter.toExternal(new SeaTunnelRow(new Object[] {value}),
statement);
+ if (value == null || value instanceof LocalDateTime) {
+ verify(statement).setObject(1, value);
+ } else {
+ verify(statement).setTimestamp(1, timestamp);
+ }
+ verifyNoMoreInteractions(statement);
+ }
+ }
+
+ @Test
+ void wrapsOnlyRecognizedTimestampParametersAndPreservesRepeatedBindings()
throws Exception {
+ String[] fields = {"id", "id2"};
+ Map<String, String> target = new HashMap<>();
+ target.put("id", "Nullable(DateTime64(6, 'America/New_York'))");
+ target.put("id2", "DateTime64(3, 'UTC')");
+ JdbcRowConverter converter =
+ new JdbcRowConverter(
+ new SeaTunnelRowType(
+ fields,
+ new SeaTunnelDataType[] {
+ LocalTimeType.OFFSET_DATE_TIME_TYPE,
+ LocalTimeType.LOCAL_DATE_TIME_TYPE
+ }),
+ target,
+ fields);
+ Connection connection = mock(Connection.class);
+ PreparedStatement delegate = mock(PreparedStatement.class);
+ String expected =
+ "SELECT 1 FROM events WHERE id = toDateTime64(?, 6, 'UTC') AND
id2 = ? OR id = toDateTime64(?, 6, 'UTC')";
+ when(connection.prepareStatement(expected)).thenReturn(delegate);
+ PreparedStatement statement =
+ converter.prepareStatement(
+ connection,
+ "SELECT 1 FROM events WHERE id = :id AND id2 = :id2 OR
id = :id",
+ fields);
+ converter.toExternal(
+ new SeaTunnelRow(
+ new Object[] {
+
OffsetDateTime.parse("2026-11-01T01:30:00.123456789-05:00"),
+ LocalDateTime.parse("2026-09-12T10:00:00")
+ }),
+ statement);
+ verify(connection).prepareStatement(expected);
+ verify(delegate).setString(1, "2026-11-01 06:30:00.123456789");
+ verify(delegate).setString(3, "2026-11-01 06:30:00.123456789");
+ verify(delegate).setObject(2,
LocalDateTime.parse("2026-09-12T10:00:00"));
+ verifyNoMoreInteractions(delegate);
+ }
+
+ @Test
+ void preservesSqlForExistingTimestampTypes() throws Exception {
+ String[] fields = {"event_time"};
+ JdbcRowConverter converter =
+ new JdbcRowConverter(
+ new SeaTunnelRowType(
+ fields,
+ new SeaTunnelDataType[]
{LocalTimeType.LOCAL_DATE_TIME_TYPE}),
+ Collections.singletonMap("event_time", "DateTime64(9,
'UTC')"),
+ fields);
+ Connection connection = mock(Connection.class);
+ converter.prepareStatement(
+ connection, SqlUtils.getInsertIntoStatement("events", fields),
fields);
+ verify(connection).prepareStatement("INSERT INTO events
(\"event_time\") VALUES (?)");
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {0, 3, 6, 9})
+ void preservesTargetDateTime64Scale(int scale) throws Exception {
+ String[] fields = {"event_time"};
+ Connection connection = mock(Connection.class);
+ converter("DateTime64(" + scale + ")")
+ .prepareStatement(
+ connection, SqlUtils.getInsertIntoStatement("events",
fields), fields);
+ verify(connection)
+ .prepareStatement(
+ "INSERT INTO events (\"event_time\") VALUES
(toDateTime64(?, "
+ + scale
+ + ", 'UTC'))");
+ }
+
+ @Test
+ void bindsNullWithoutFormattingIt() throws Exception {
+ PreparedStatement statement = mock(PreparedStatement.class);
+ converter("Nullable(DateTime64(3, 'UTC'))")
+ .toExternal(new SeaTunnelRow(new Object[] {null}), statement);
+ verify(statement).setObject(1, null);
+ verifyNoMoreInteractions(statement);
+ }
+
+ @Test
+ void rejectsPositionalTimestampParametersWithoutUtcConversion() {
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ converter("DateTime64(9, 'UTC')")
+ .prepareStatement(
+ mock(Connection.class),
+ "INSERT INTO events VALUES (?)",
+ new String[] {"event_time"}));
+ }
+
+ private JdbcRowConverter converter(String targetType) {
+ return new JdbcRowConverter(
+ new SeaTunnelRowType(
+ new String[] {"event_time"},
+ new SeaTunnelDataType[]
{LocalTimeType.OFFSET_DATE_TIME_TYPE}),
+ Collections.singletonMap("event_time", targetType),
+ new String[] {"event_time"});
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseOffsetTimestampIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseOffsetTimestampIT.java
new file mode 100644
index 0000000000..2f8123bdfe
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseOffsetTimestampIT.java
@@ -0,0 +1,269 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.clickhouse;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.sink.SaveModeHandler;
+import org.apache.seatunnel.api.sink.SinkWriter;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext;
+import org.apache.seatunnel.api.table.type.ArrayType;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.DecimalType;
+import org.apache.seatunnel.api.table.type.LocalTimeType;
+import org.apache.seatunnel.api.table.type.RowKind;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import
org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.client.ClickhouseSink;
+import
org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.client.ClickhouseSinkFactory;
+import
org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.client.ClickhouseSinkWriter;
+import
org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.inject.DateTimeInjectFunction;
+
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.ClickHouseContainer;
+
+import java.math.BigDecimal;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.OffsetDateTime;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+class ClickhouseOffsetTimestampIT {
+
+ @Test
+ void writesOffsetTimestampsToExistingTable() throws Exception {
+ try (ClickHouseContainer database =
+ new
ClickHouseContainer("clickhouse/clickhouse-server:23.3.13.6")) {
+ database.start();
+ try (Connection connection =
+ DriverManager.getConnection(
+ database.getJdbcUrl(),
+ database.getUsername(),
+ database.getPassword());
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ "CREATE TABLE offset_timestamps (id Int32, seconds
DateTime('Asia/Shanghai'), micros DateTime64(6, 'America/New_York'), nanos
Nullable(DateTime64(9, 'UTC')), label String, amount Decimal(20, 6), tags
Array(String), local_time DateTime64(9, 'UTC'), sql_timestamp DateTime64(9,
'UTC')) ENGINE = MergeTree ORDER BY tuple()");
+ statement.execute(
+ "CREATE TABLE legacy_timestamp (value DateTime64(9,
'UTC')) ENGINE=Memory");
+ Properties properties = new Properties();
+ properties.setProperty("user", database.getUsername());
+ properties.setProperty("password", database.getPassword());
+ // The unwrapped insert retains the driver's existing
Timestamp conversion.
+ try (Connection legacyConnection =
+ new com.clickhouse.jdbc.ClickHouseDriver()
+ .connect(database.getJdbcUrl(),
properties);
+ PreparedStatement legacy =
+ legacyConnection.prepareStatement(
+ "INSERT INTO legacy_timestamp VALUES
(?)")) {
+ new DateTimeInjectFunction()
+ .injectFields(
+ legacy,
+ 1,
+ Timestamp.valueOf(
+
LocalDateTime.parse("2026-09-12T10:00:00.123456789")));
+ legacy.addBatch();
+ legacy.executeBatch();
+ }
+ String legacyTimestamp;
+ try (ResultSet rows =
+ statement.executeQuery("SELECT toString(value) FROM
legacy_timestamp")) {
+ assertTrue(rows.next());
+ legacyTimestamp = rows.getString(1);
+ }
+ Map<String, Object> options = new HashMap<>();
+ options.put("host", database.getHost() + ":" +
database.getMappedPort(8123));
+ options.put("database", "default");
+ options.put("table", "offset_timestamps");
+ options.put("username", database.getUsername());
+ options.put("password", database.getPassword());
+ TableSchema.Builder schema =
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.of(
+ "id", BasicType.INT_TYPE, 0L,
false, null, null));
+ for (String name : new String[] {"seconds", "micros",
"nanos"}) {
+ schema.column(
+ PhysicalColumn.of(
+ name,
+ LocalTimeType.OFFSET_DATE_TIME_TYPE,
+ 0L,
+ true,
+ null,
+ null));
+ }
+ schema.column(
+ PhysicalColumn.of("label", BasicType.STRING_TYPE, 0L,
false, null, null));
+ schema.column(
+ PhysicalColumn.of("amount", new DecimalType(20, 6),
0L, false, null, null));
+ schema.column(
+ PhysicalColumn.of(
+ "tags", ArrayType.STRING_ARRAY_TYPE, 0L,
false, null, null));
+ schema.column(
+ PhysicalColumn.of(
+ "local_time",
+ LocalTimeType.LOCAL_DATE_TIME_TYPE,
+ 0L,
+ false,
+ null,
+ null));
+ schema.column(
+ PhysicalColumn.of(
+ "sql_timestamp",
+ LocalTimeType.LOCAL_DATE_TIME_TYPE,
+ 0L,
+ false,
+ null,
+ null));
+ CatalogTable table =
+ CatalogTable.of(
+ TableIdentifier.of("clickhouse", "default",
"offset_timestamps"),
+ schema.build(),
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ null);
+ ClickhouseSink sink =
+ (ClickhouseSink)
+ new ClickhouseSinkFactory()
+ .createSink(
+ new TableSinkFactoryContext(
+ table,
+
ReadonlyConfig.fromMap(options),
+
getClass().getClassLoader()))
+ .createSink();
+ try (SaveModeHandler saveMode =
sink.getSaveModeHandler().get()) {
+ saveMode.open();
+ saveMode.handleSaveMode();
+ }
+ String[] literals = {
+ "2026-09-12T10:00:00.123456789Z",
+ "2026-09-12T10:00:00.123456789+05:30",
+ "2026-09-12T10:00:00.123456789-07:00",
+ "2026-11-01T01:30:00.123456789-04:00",
+ "2026-11-01T01:30:00.123456789-05:00"
+ };
+ ClickhouseSinkWriter writer =
sink.createWriter(mock(SinkWriter.Context.class));
+ try {
+ for (int i = 0; i < literals.length; i++) {
+ OffsetDateTime value =
OffsetDateTime.parse(literals[i]);
+ writer.write(row(i, value, value));
+ }
+ OffsetDateTime value = OffsetDateTime.parse(literals[0]);
+ writer.write(row(literals.length, value, null));
+ writer.prepareCommit();
+ } finally {
+ writer.close();
+ }
+ try (ResultSet rows =
+ statement.executeQuery(
+ "SELECT id, toUnixTimestamp(seconds),
toUnixTimestamp64Micro(micros), toUnixTimestamp64Nano(nanos), isNull(nanos),
label, amount, tags, toString(local_time), toString(sql_timestamp) FROM
offset_timestamps ORDER BY id")) {
+ for (int i = 0; i < literals.length; i++) {
+ assertTrue(rows.next());
+ Instant expected =
OffsetDateTime.parse(literals[i]).toInstant();
+ assertEquals(i, rows.getInt(1));
+ assertEquals(expected.getEpochSecond(),
rows.getLong(2));
+ assertEquals(
+ expected.getEpochSecond() * 1_000_000L +
expected.getNano() / 1_000,
+ rows.getLong(3));
+ assertEquals(
+ expected.getEpochSecond() * 1_000_000_000L +
expected.getNano(),
+ rows.getLong(4));
+ assertEquals("a'b\\c", rows.getString(6));
+ assertEquals(new BigDecimal("1234567890.123456"),
rows.getBigDecimal(7));
+ assertArrayEquals(
+ new String[] {"a'b", "c\\d"},
+ (Object[]) rows.getArray(8).getArray());
+ assertEquals("2026-09-12 10:00:00.123456789",
rows.getString(9));
+ assertEquals(legacyTimestamp, rows.getString(10));
+ }
+ assertTrue(rows.next());
+ assertEquals(literals.length, rows.getInt(1));
+ assertTrue(rows.getBoolean(5));
+ assertFalse(rows.next());
+ }
+ options.put("primary_key", "micros");
+ options.put("support_upsert", true);
+ ClickhouseSink upsert =
+ (ClickhouseSink)
+ new ClickhouseSinkFactory()
+ .createSink(
+ new TableSinkFactoryContext(
+ table,
+
ReadonlyConfig.fromMap(options),
+
getClass().getClassLoader()))
+ .createSink();
+ writer = upsert.createWriter(mock(SinkWriter.Context.class));
+ try {
+ // Both overlap occurrences have the same local time in
the target column.
+ SeaTunnelRow updated =
+ row(10,
OffsetDateTime.parse("2026-11-01T05:30:00.123456789Z"), null);
+ updated.setRowKind(RowKind.UPDATE_AFTER);
+ writer.write(updated);
+ writer.prepareCommit();
+ SeaTunnelRow deleted =
+ row(4,
OffsetDateTime.parse("2026-11-01T06:30:00.123456789Z"), null);
+ deleted.setRowKind(RowKind.DELETE);
+ writer.write(deleted);
+ writer.prepareCommit();
+ } finally {
+ writer.close();
+ }
+ try (ResultSet rows =
+ statement.executeQuery("SELECT id FROM
offset_timestamps ORDER BY id")) {
+ for (int id : new int[] {0, 1, 2, 5, 10}) {
+ assertTrue(rows.next());
+ assertEquals(id, rows.getInt(1));
+ }
+ assertFalse(rows.next());
+ }
+ }
+ }
+ }
+
+ private SeaTunnelRow row(int id, OffsetDateTime timestamp, OffsetDateTime
nullableTimestamp) {
+ LocalDateTime local =
LocalDateTime.parse("2026-09-12T10:00:00.123456789");
+ return new SeaTunnelRow(
+ new Object[] {
+ id,
+ timestamp,
+ timestamp,
+ nullableTimestamp,
+ "a'b\\c",
+ new BigDecimal("1234567890.123456"),
+ new String[] {"a'b", "c\\d"},
+ local,
+ Timestamp.valueOf(local)
+ });
+ }
+}
