This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12285-e486ac9deda488259f176fedd9498d2a901f6246 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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) + }); + } +}
