You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ku...@apache.org on 2026/09/26 01:56:26 UTC
(flink-cdc) branch master updated: [FLINK-38835][postgres] Fix timestamp conversion to LocalDateTime for dates before 1970-01-01. (#4214)
This is an automated email from the ASF dual-hosted git repository.
lvyanquan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 6599fc242 [FLINK-38835][postgres] Fix timestamp conversion to LocalDateTime for dates before 1970-01-01. (#4214)
6599fc242 is described below
commit 6599fc242b04598425e76287ef9d663bb6e04dce
Author: Kunni <de...@gmail.com>
AuthorDate: Sat Sep 26 09:56:18 2026 +0800
[FLINK-38835][postgres] Fix timestamp conversion to LocalDateTime for dates before 1970-01-01. (#4214)
---
.../docs/connectors/flink-sources/postgres-cdc.md | 15 ++
.../connectors/pipeline-connectors/postgres.md | 13 ++
.../docs/connectors/flink-sources/postgres-cdc.md | 19 +++
.../connectors/pipeline-connectors/postgres.md | 15 ++
.../factory/PostgresDataSourceFactory.java | 25 ++-
.../postgres/source/PostgresDataSourceOptions.java | 17 ++
.../factory/PostgresDataSourceFactoryTest.java | 33 ++++
.../postgresql/CustomPostgresValueConverter.java | 137 ++++++++++++++++
.../connector/postgresql/PostgresObjectUtils.java | 2 +-
.../postgresql/connection/PostgresConnection.java | 4 +-
.../source/config/PostgresSourceOptions.java | 21 +++
.../postgres/table/PostgreSQLTableFactory.java | 19 ++-
.../CustomPostgresValueConverterTest.java | 182 +++++++++++++++++++++
.../postgres/table/PostgreSQLConnectorITCase.java | 168 +++++++++++++++++++
14 files changed, 666 insertions(+), 4 deletions(-)
diff --git a/docs/content.zh/docs/connectors/flink-sources/postgres-cdc.md b/docs/content.zh/docs/connectors/flink-sources/postgres-cdc.md
index 78e662f3a..01baf131a 100644
--- a/docs/content.zh/docs/connectors/flink-sources/postgres-cdc.md
+++ b/docs/content.zh/docs/connectors/flink-sources/postgres-cdc.md
@@ -299,6 +299,21 @@ Connector Options
(2) Table list (regex or predefined list) should only match the parent table name, if table list matches both parent and child tables, snapshot data will be read twice.
</td>
</tr>
+ <tr>
+ <td>scan.pre-epoch-timestamp.wall-clock-conversion.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>
+ 是否对 1970-01-01（epoch）之前的 PostgreSQL "timestamp without time zone" 列值，按照数据库中存储的日期时间（wall clock）进行转换。<br>
+ 开启后，该值不会受运行任务的 JVM 时区影响，这在存在历史时区偏移的时区下很重要：当 JVM 时区为 Asia/Shanghai 时，
+ 数据库中存储的 "1900-01-01 00:00:00.123" 否则会被读取为 "1900-01-01 00:05:43.123"。<br>
+ 关闭（默认值）时保持原有的转换行为。1970-01-01 及之后的值无论是否开启该选项都会被同样地转换，
+ 且该选项不会改变列的数据类型。<br>
+ 该选项仅在 <code>scan.incremental.snapshot.enabled</code> 为 true 时生效，因为非增量快照源的数据值由 Debezium 自身转换。<br>
+ 使用 DataStream API 时，可以通过在 Debezium 参数中添加 "scan.pre-epoch-timestamp.wall-clock-conversion.enabled" 来开启该行为。
+ </td>
+ </tr>
</tbody>
</table>
</div>
diff --git a/docs/content.zh/docs/connectors/pipeline-connectors/postgres.md b/docs/content.zh/docs/connectors/pipeline-connectors/postgres.md
index 4a231d4ec..146346ba7 100644
--- a/docs/content.zh/docs/connectors/pipeline-connectors/postgres.md
+++ b/docs/content.zh/docs/connectors/pipeline-connectors/postgres.md
@@ -295,6 +295,19 @@ pipeline:
以发现在 savepoint 时刻未被捕获但匹配 source <code>tables</code> 模式的新表。
</td>
</tr>
+ <tr>
+ <td>scan.pre-epoch-timestamp.wall-clock-conversion.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>
+ 是否对 1970-01-01（epoch）之前的 PostgreSQL "timestamp without time zone" 列值，按照数据库中存储的日期时间（wall clock）进行转换。<br>
+ 开启后，该值不会受运行任务的 JVM 时区影响，这在存在历史时区偏移的时区下很重要：当 JVM 时区为 Asia/Shanghai 时，
+ 数据库中存储的 "1900-01-01 00:00:00.123" 否则会被读取为 "1900-01-01 00:05:43.123"。<br>
+ 关闭（默认值）时保持原有的转换行为。1970-01-01 及之后的值无论是否开启该选项都会被同样地转换，
+ 且该选项不会改变列的数据类型。
+ </td>
+ </tr>
<tr>
<td>schema-change.enabled</td>
<td>optional</td>
diff --git a/docs/content/docs/connectors/flink-sources/postgres-cdc.md b/docs/content/docs/connectors/flink-sources/postgres-cdc.md
index 6f507e1b0..0eafb4fb2 100644
--- a/docs/content/docs/connectors/flink-sources/postgres-cdc.md
+++ b/docs/content/docs/connectors/flink-sources/postgres-cdc.md
@@ -296,6 +296,25 @@ SELECT * FROM shipments;
(2) Table list (regex or predefined list) should only match the parent table name, if table list matches both parent and child tables, snapshot data will be read twice.
</td>
</tr>
+ <tr>
+ <td>scan.pre-epoch-timestamp.wall-clock-conversion.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>
+ Whether to convert the values of PostgreSQL "timestamp without time zone" columns which are before 1970-01-01 (the epoch)
+ by keeping the date and time (wall clock) stored in the database.<br>
+ If enabled, such a value is not changed by the time zone of the JVM running the job, which matters in time zones which have
+ historical offsets: with the JVM time zone set to Asia/Shanghai, the stored value "1900-01-01 00:00:00.123" would otherwise
+ be read as "1900-01-01 00:05:43.123".<br>
+ If disabled (default), the previous conversion behavior is kept. Values which are not before 1970-01-01 are converted in the
+ same way whatever the value of this option is, and this option never changes the data type of a column.<br>
+ This option only takes effect when <code>scan.incremental.snapshot.enabled</code> is true, because the values of the
+ non-incremental snapshot source are converted by Debezium itself.<br>
+ When the source is created with the DataStream API, this behavior is enabled by adding the
+ "scan.pre-epoch-timestamp.wall-clock-conversion.enabled" property to the Debezium properties.
+ </td>
+ </tr>
</tbody>
</table>
</div>
diff --git a/docs/content/docs/connectors/pipeline-connectors/postgres.md b/docs/content/docs/connectors/pipeline-connectors/postgres.md
index 174261258..189ce87c7 100644
--- a/docs/content/docs/connectors/pipeline-connectors/postgres.md
+++ b/docs/content/docs/connectors/pipeline-connectors/postgres.md
@@ -288,6 +288,21 @@ pipeline:
the captured set at savepoint time.
</td>
</tr>
+ <tr>
+ <td>scan.pre-epoch-timestamp.wall-clock-conversion.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>
+ Whether to convert the values of PostgreSQL "timestamp without time zone" columns which are before 1970-01-01 (the epoch)
+ by keeping the date and time (wall clock) stored in the database.<br>
+ If enabled, such a value is not changed by the time zone of the JVM running the job, which matters in time zones which have
+ historical offsets: with the JVM time zone set to Asia/Shanghai, the stored value "1900-01-01 00:00:00.123" would otherwise
+ be read as "1900-01-01 00:05:43.123".<br>
+ If disabled (default), the previous conversion behavior is kept. Values which are not before 1970-01-01 are converted in the
+ same way whatever the value of this option is, and this option never changes the data type of a column.
+ </td>
+ </tr>
<tr>
<td>schema-change.enabled</td>
<td>optional</td>
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactory.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactory.java
index 5a7f5f932..45e4accc9 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactory.java
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactory.java
@@ -50,6 +50,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Properties;
import java.util.Set;
import java.util.stream.Collectors;
@@ -73,6 +74,7 @@ import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSource
import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED;
import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SCAN_LSN_COMMIT_CHECKPOINTS_DELAY;
import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED;
+import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SCAN_PRE_EPOCH_TIMESTAMP_WALL_CLOCK_CONVERSION_ENABLED;
import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SCAN_SNAPSHOT_FETCH_SIZE;
import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SCAN_STARTUP_MODE;
import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SCHEMA_CHANGE_ENABLED;
@@ -159,7 +161,7 @@ public class PostgresDataSourceFactory implements DataSourceFactory {
.decodingPluginName(pluginName)
.slotName(slotName)
.serverTimeZone(serverTimeZone.getId())
- .debeziumProperties(getDebeziumProperties(configMap))
+ .debeziumProperties(getDbzProperties(config, configMap))
.splitSize(splitSize)
.splitMetaGroupSize(splitMetaGroupSize)
.distributionFactorUpper(distributionFactorUpper)
@@ -234,6 +236,26 @@ public class PostgresDataSourceFactory implements DataSourceFactory {
String.join(", ", readableMetadataList)));
}
+ /**
+ * Returns the Debezium properties of the pipeline source. The timestamp conversion is done by
+ * the Debezium value converter, so {@link
+ * #SCAN_PRE_EPOCH_TIMESTAMP_WALL_CLOCK_CONVERSION_ENABLED} is forwarded as a Debezium property.
+ * It is only set when it's configured explicitly to keep a property configured via
+ * 'debezium.scan.pre-epoch-timestamp.wall-clock-conversion.enabled' effective.
+ */
+ private static Properties getDbzProperties(
+ Configuration config, Map<String, String> configMap) {
+ Properties dbzProperties = getDebeziumProperties(configMap);
+ config.getOptional(SCAN_PRE_EPOCH_TIMESTAMP_WALL_CLOCK_CONVERSION_ENABLED)
+ .ifPresent(
+ enabled ->
+ dbzProperties.setProperty(
+ SCAN_PRE_EPOCH_TIMESTAMP_WALL_CLOCK_CONVERSION_ENABLED
+ .key(),
+ String.valueOf(enabled)));
+ return dbzProperties;
+ }
+
@Override
public Set<ConfigOption<?>> requiredOptions() {
Set<ConfigOption<?>> options = new HashSet<>();
@@ -270,6 +292,7 @@ public class PostgresDataSourceFactory implements DataSourceFactory {
options.add(TABLE_ID_INCLUDE_DATABASE);
options.add(SCHEMA_CHANGE_ENABLED);
options.add(SCAN_NEWLY_ADDED_TABLE_ENABLED);
+ options.add(SCAN_PRE_EPOCH_TIMESTAMP_WALL_CLOCK_CONVERSION_ENABLED);
return options;
}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSourceOptions.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSourceOptions.java
index 183d7cc89..55859fe35 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSourceOptions.java
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDataSourceOptions.java
@@ -274,6 +274,23 @@ public class PostgresDataSourceOptions {
+ "If set to true, the Table ID will be in the format (database, schema, table). "
+ "If set to false, the Table ID will be in the format (schema, table). Defaults to false.");
+ public static final ConfigOption<Boolean>
+ SCAN_PRE_EPOCH_TIMESTAMP_WALL_CLOCK_CONVERSION_ENABLED =
+ ConfigOptions.key("scan.pre-epoch-timestamp.wall-clock-conversion.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to convert the values of PostgreSQL \"timestamp without time zone\" columns which are "
+ + "before 1970-01-01 (the epoch) by keeping the date and time (wall clock) stored in the "
+ + "database.\n"
+ + "If set to true, such a value is not changed by the time zone of the JVM running the job. "
+ + "This matters in time zones which have historical offsets: with the JVM time zone set to "
+ + "Asia/Shanghai, the stored value \"1900-01-01 00:00:00.123\" would otherwise be read as "
+ + "\"1900-01-01 00:05:43.123\".\n"
+ + "If set to false (default), the previous conversion behavior is kept. Values which are "
+ + "not before 1970-01-01 are converted in the same way whatever the value of this option "
+ + "is, and this option never changes the data type of a column.");
+
@Experimental
public static final ConfigOption<Boolean> SCHEMA_CHANGE_ENABLED =
ConfigOptions.key("schema-change.enabled")
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactoryTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactoryTest.java
index f64a0935b..24405879e 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactoryTest.java
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/factory/PostgresDataSourceFactoryTest.java
@@ -49,6 +49,7 @@ import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSource
import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.PASSWORD;
import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.PG_PORT;
import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED;
+import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SCAN_PRE_EPOCH_TIMESTAMP_WALL_CLOCK_CONVERSION_ENABLED;
import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.SLOT_NAME;
import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.TABLES;
import static org.apache.flink.cdc.connectors.postgres.source.PostgresDataSourceOptions.TABLES_EXCLUDE;
@@ -262,6 +263,38 @@ public class PostgresDataSourceFactoryTest extends PostgresTestBase {
.isEqualTo(POSTGRES_CONTAINER.getMappedPort(POSTGRESQL_PORT));
}
+ @Test
+ public void testWallClockConversionOption() {
+ Map<String, String> options = new HashMap<>();
+ options.put(HOSTNAME.key(), POSTGRES_CONTAINER.getHost());
+ options.put(
+ PG_PORT.key(), String.valueOf(POSTGRES_CONTAINER.getMappedPort(POSTGRESQL_PORT)));
+ options.put(USERNAME.key(), TEST_USER);
+ options.put(PASSWORD.key(), TEST_PASSWORD);
+ options.put(TABLES.key(), POSTGRES_CONTAINER.getDatabaseName() + ".inventory.prod\\.*");
+ options.put(SLOT_NAME.key(), slotName);
+ options.put(SCAN_PRE_EPOCH_TIMESTAMP_WALL_CLOCK_CONVERSION_ENABLED.key(), "true");
+
+ Factory.Context context = new MockContext(Configuration.fromMap(options));
+ PostgresDataSourceFactory factory = new PostgresDataSourceFactory();
+ assertThat(factory.optionalOptions())
+ .contains(SCAN_PRE_EPOCH_TIMESTAMP_WALL_CLOCK_CONVERSION_ENABLED);
+
+ PostgresDataSource dataSource = (PostgresDataSource) factory.createDataSource(context);
+ // the option is forwarded to the Debezium configuration which reads it by the value
+ // converter
+ assertThat(
+ dataSource
+ .getPostgresSourceConfig()
+ .getDbzConnectorConfig()
+ .getConfig()
+ .getBoolean(
+ SCAN_PRE_EPOCH_TIMESTAMP_WALL_CLOCK_CONVERSION_ENABLED
+ .key(),
+ false))
+ .isTrue();
+ }
+
@Test
public void testPrefixRequireOption() {
Map<String, String> options = new HashMap<>();
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/CustomPostgresValueConverter.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/CustomPostgresValueConverter.java
new file mode 100644
index 000000000..29118666e
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/CustomPostgresValueConverter.java
@@ -0,0 +1,137 @@
+/*
+ * 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 io.debezium.connector.postgresql;
+
+import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions;
+
+import io.debezium.config.CommonConnectorConfig;
+import io.debezium.config.Configuration;
+import io.debezium.jdbc.TemporalPrecisionMode;
+import io.debezium.relational.Column;
+import org.apache.kafka.connect.data.Field;
+
+import java.nio.charset.Charset;
+import java.sql.Timestamp;
+import java.time.ZoneOffset;
+
+/**
+ * A {@link PostgresValueConverter} which is able to keep the date and time (wall clock) stored in
+ * PostgreSQL for {@code timestamp} values, instead of letting the time zone of the JVM shift them.
+ *
+ * <p>This behavior is only enabled by {@link
+ * PostgresSourceOptions#SCAN_PRE_EPOCH_TIMESTAMP_WALL_CLOCK_CONVERSION_ENABLED}, otherwise the
+ * conversion of the parent implementation is kept. Only {@code timestamp} (without time zone)
+ * values are affected, timestamp values with time zone always use the parent implementation, and
+ * the schema of the converted values is never changed. The constructor and factory mirror the
+ * Debezium implementation and must remain aligned with it when the Debezium version changes.
+ */
+public class CustomPostgresValueConverter extends PostgresValueConverter {
+
+ private final boolean wallClockConversionEnabled;
+
+ protected CustomPostgresValueConverter(
+ Charset databaseCharset,
+ DecimalMode decimalMode,
+ TemporalPrecisionMode temporalPrecisionMode,
+ ZoneOffset defaultOffset,
+ BigIntUnsignedMode bigIntUnsignedMode,
+ boolean includeUnknownDatatypes,
+ TypeRegistry typeRegistry,
+ PostgresConnectorConfig.HStoreHandlingMode hStoreMode,
+ CommonConnectorConfig.BinaryHandlingMode binaryMode,
+ PostgresConnectorConfig.IntervalHandlingMode intervalMode,
+ byte[] toastPlaceholder,
+ int moneyFractionDigits,
+ boolean wallClockConversionEnabled) {
+ super(
+ databaseCharset,
+ decimalMode,
+ temporalPrecisionMode,
+ defaultOffset,
+ bigIntUnsignedMode,
+ includeUnknownDatatypes,
+ typeRegistry,
+ hStoreMode,
+ binaryMode,
+ intervalMode,
+ toastPlaceholder,
+ moneyFractionDigits);
+ this.wallClockConversionEnabled = wallClockConversionEnabled;
+ }
+
+ public static CustomPostgresValueConverter of(
+ PostgresConnectorConfig connectorConfig,
+ Charset databaseCharset,
+ TypeRegistry typeRegistry) {
+ Configuration config = connectorConfig.getConfig();
+ boolean wallClockConversionEnabled =
+ config.getBoolean(
+ PostgresSourceOptions.SCAN_PRE_EPOCH_TIMESTAMP_WALL_CLOCK_CONVERSION_ENABLED
+ .key(),
+ PostgresSourceOptions.SCAN_PRE_EPOCH_TIMESTAMP_WALL_CLOCK_CONVERSION_ENABLED
+ .defaultValue());
+ return new CustomPostgresValueConverter(
+ databaseCharset,
+ connectorConfig.getDecimalMode(),
+ connectorConfig.getTemporalPrecisionMode(),
+ ZoneOffset.UTC,
+ null,
+ connectorConfig.includeUnknownDatatypes(),
+ typeRegistry,
+ connectorConfig.hStoreHandlingMode(),
+ connectorConfig.binaryHandlingMode(),
+ connectorConfig.intervalHandlingMode(),
+ connectorConfig.getUnavailableValuePlaceholder(),
+ connectorConfig.moneyFractionDigits(),
+ wallClockConversionEnabled);
+ }
+
+ /**
+ * Converts a PostgreSQL {@code timestamp} to {@link java.time.LocalDateTime} with the date and
+ * time (wall clock) stored in the database, if {@link
+ * PostgresSourceOptions#SCAN_PRE_EPOCH_TIMESTAMP_WALL_CLOCK_CONVERSION_ENABLED} is enabled.
+ *
+ * <p>The parent implementation converts through {@link Timestamp#toInstant()}, which applies
+ * the historical offsets of the JVM default time zone and may thus shift the date and time
+ * fields of values before 1970-01-01. The column and field definitions are intentionally unused
+ * because this conversion only depends on the JDBC value.
+ */
+ @Override
+ protected Object convertTimestampToLocalDateTime(Column column, Field fieldDefn, Object data) {
+ if (!wallClockConversionEnabled) {
+ // Keep the behavior of Debezium for backward compatibility.
+ return super.convertTimestampToLocalDateTime(column, fieldDefn, data);
+ }
+ if (data == null) {
+ return null;
+ }
+ if (!(data instanceof Timestamp)) {
+ return data;
+ }
+ final Timestamp timestamp = (Timestamp) data;
+
+ if (POSITIVE_INFINITY_TIMESTAMP.equals(timestamp)) {
+ return POSITIVE_INFINITY_LOCAL_DATE_TIME;
+ } else if (NEGATIVE_INFINITY_TIMESTAMP.equals(timestamp)) {
+ return NEGATIVE_INFINITY_LOCAL_DATE_TIME;
+ }
+
+ // Preserve the timestamp's wall-clock fields instead of applying historical zone offsets.
+ return timestamp.toLocalDateTime();
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/PostgresObjectUtils.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/PostgresObjectUtils.java
index f5132bffd..4aec2dae7 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/PostgresObjectUtils.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/PostgresObjectUtils.java
@@ -103,7 +103,7 @@ public class PostgresObjectUtils {
public static PostgresConnection.PostgresValueConverterBuilder newPostgresValueConverterBuilder(
PostgresConnectorConfig config) {
return typeRegistry ->
- PostgresValueConverter.of(config, StandardCharsets.UTF_8, typeRegistry);
+ CustomPostgresValueConverter.of(config, StandardCharsets.UTF_8, typeRegistry);
}
// modified from
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/connection/PostgresConnection.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/connection/PostgresConnection.java
index 610bd4360..f6f286c27 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/connection/PostgresConnection.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/connection/PostgresConnection.java
@@ -10,6 +10,7 @@ import com.zaxxer.hikari.pool.HikariProxyConnection;
import io.debezium.DebeziumException;
import io.debezium.annotation.VisibleForTesting;
import io.debezium.config.Configuration;
+import io.debezium.connector.postgresql.CustomPostgresValueConverter;
import io.debezium.connector.postgresql.PgOid;
import io.debezium.connector.postgresql.PostgresConnectorConfig;
import io.debezium.connector.postgresql.PostgresSchema;
@@ -195,7 +196,8 @@ public class PostgresConnection extends JdbcConnection {
} else {
this.typeRegistry = typeRegistry;
final PostgresValueConverter valueConverter =
- PostgresValueConverter.of(config, this.getDatabaseCharset(), typeRegistry);
+ CustomPostgresValueConverter.of(
+ config, this.getDatabaseCharset(), typeRegistry);
this.defaultValueConverter =
new PostgresDefaultValueConverter(valueConverter, this.getTimestampUtils());
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceOptions.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceOptions.java
index db04ac142..ecf55808f 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceOptions.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceOptions.java
@@ -106,4 +106,25 @@ public class PostgresSourceOptions extends JdbcSourceOptions {
"Whether to include database in the generated Table ID.\n"
+ "If set to true, the Table ID will be in the format (database, schema, table).\n"
+ "If set to false, the Table ID will be in the format (schema, table). Defaults to false.");
+
+ public static final ConfigOption<Boolean>
+ SCAN_PRE_EPOCH_TIMESTAMP_WALL_CLOCK_CONVERSION_ENABLED =
+ ConfigOptions.key("scan.pre-epoch-timestamp.wall-clock-conversion.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to convert the values of PostgreSQL \"timestamp without time zone\" columns which are "
+ + "before 1970-01-01 (the epoch) by keeping the date and time (wall clock) stored in the "
+ + "database.\n"
+ + "If set to true, such a value is not changed by the time zone of the JVM running the job. "
+ + "This matters in time zones which have historical offsets: with the JVM time zone set to "
+ + "Asia/Shanghai, the stored value \"1900-01-01 00:00:00.123\" would otherwise be read as "
+ + "\"1900-01-01 00:05:43.123\".\n"
+ + "If set to false (default), the previous conversion behavior is kept.\n"
+ + "Values which are not before 1970-01-01 are converted in the same way whatever the value "
+ + "of this option is, and this option never changes the data type of a column.\n"
+ + "This option only takes effect when 'scan.incremental.snapshot.enabled' is true, because "
+ + "the values of the non-incremental snapshot source are converted by Debezium itself.\n"
+ + "When the source is created with the DataStream API, this behavior is enabled by adding "
+ + "this option key to the Debezium properties.");
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactory.java
index 971a9d4cd..af40d1de6 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactory.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactory.java
@@ -33,6 +33,7 @@ import org.apache.flink.table.factories.FactoryUtil;
import java.time.Duration;
import java.util.HashSet;
+import java.util.Properties;
import java.util.Set;
import static org.apache.flink.cdc.connectors.base.options.JdbcSourceOptions.DATABASE_NAME;
@@ -61,6 +62,7 @@ import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSou
import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE;
import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ENABLED;
import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions.SCAN_LSN_COMMIT_CHECKPOINTS_DELAY;
+import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions.SCAN_PRE_EPOCH_TIMESTAMP_WALL_CLOCK_CONVERSION_ENABLED;
import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions.SCAN_SNAPSHOT_FETCH_SIZE;
import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions.SCAN_STARTUP_MODE;
import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions.SLOT_NAME;
@@ -126,6 +128,20 @@ public class PostgreSQLTableFactory implements DynamicTableSourceFactory {
boolean releaseSnapshotMetadataEnabled =
config.get(SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED);
+ Properties dbzProperties = getDebeziumProperties(context.getCatalogTable().getOptions());
+ // The timestamp conversion is done by the Debezium value converter, so the option is
+ // forwarded as a Debezium property. It is only set when it's configured explicitly to keep
+ // a property configured via
+ // 'debezium.scan.pre-epoch-timestamp.wall-clock-conversion.enabled'
+ // effective.
+ config.getOptional(SCAN_PRE_EPOCH_TIMESTAMP_WALL_CLOCK_CONVERSION_ENABLED)
+ .ifPresent(
+ enabled ->
+ dbzProperties.setProperty(
+ SCAN_PRE_EPOCH_TIMESTAMP_WALL_CLOCK_CONVERSION_ENABLED
+ .key(),
+ String.valueOf(enabled)));
+
if (enableParallelRead) {
validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1);
validateIntegerOption(SCAN_SNAPSHOT_FETCH_SIZE, fetchSize, 1);
@@ -154,7 +170,7 @@ public class PostgreSQLTableFactory implements DynamicTableSourceFactory {
pluginName,
slotName,
changelogMode,
- getDebeziumProperties(context.getCatalogTable().getOptions()),
+ dbzProperties,
enableParallelRead,
splitSize,
splitMetaGroupSize,
@@ -221,6 +237,7 @@ public class PostgreSQLTableFactory implements DynamicTableSourceFactory {
options.add(SCAN_READ_CHANGELOG_AS_APPEND_ONLY_ENABLED);
options.add(SCAN_INCLUDE_PARTITIONED_TABLES_ENABLED);
options.add(SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED);
+ options.add(SCAN_PRE_EPOCH_TIMESTAMP_WALL_CLOCK_CONVERSION_ENABLED);
return options;
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/io/debezium/connector/postgresql/CustomPostgresValueConverterTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/io/debezium/connector/postgresql/CustomPostgresValueConverterTest.java
new file mode 100644
index 000000000..2ab1f6e2b
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/io/debezium/connector/postgresql/CustomPostgresValueConverterTest.java
@@ -0,0 +1,182 @@
+/*
+ * 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 io.debezium.connector.postgresql;
+
+import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfig;
+import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfigFactory;
+import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions;
+import org.apache.flink.cdc.connectors.postgres.testutils.TestHelper;
+
+import io.debezium.config.Configuration;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.util.Properties;
+import java.util.TimeZone;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class CustomPostgresValueConverterTest {
+
+ private static TimeZone originalTimeZone;
+
+ private final CustomPostgresValueConverter wallClockConverter = customConverter(true);
+ private final CustomPostgresValueConverter defaultConverter = customConverter(false);
+ private final PostgresValueConverter debeziumConverter = debeziumConverter();
+
+ @BeforeAll
+ static void useRegionalTimeZone() {
+ originalTimeZone = TimeZone.getDefault();
+ TimeZone.setDefault(TimeZone.getTimeZone("Asia/Shanghai"));
+ }
+
+ @AfterAll
+ static void restoreTimeZone() {
+ TimeZone.setDefault(originalTimeZone);
+ }
+
+ @Test
+ void testPreEpochTimestampKeepsWallClockWhenEnabled() {
+ assertThat(convert(wallClockConverter, "1600-01-01 00:00:00.123456"))
+ .isEqualTo(LocalDateTime.parse("1600-01-01T00:00:00.123456"));
+ assertThat(convert(wallClockConverter, "1900-01-01 00:00:00.123456"))
+ .isEqualTo(LocalDateTime.parse("1900-01-01T00:00:00.123456"));
+ }
+
+ @Test
+ void testTimestampAroundEpochKeepsWallClockWhenEnabled() {
+ assertThat(convert(wallClockConverter, "1969-12-31 23:59:59.999999"))
+ .isEqualTo(LocalDateTime.parse("1969-12-31T23:59:59.999999"));
+ assertThat(convert(wallClockConverter, "1970-01-01 00:00:00"))
+ .isEqualTo(LocalDateTime.parse("1970-01-01T00:00:00"));
+ assertThat(convert(wallClockConverter, "2020-07-17 18:00:22.123456"))
+ .isEqualTo(LocalDateTime.parse("2020-07-17T18:00:22.123456"));
+ }
+
+ @Test
+ void testDefaultKeepsDebeziumConversion() {
+ for (String value :
+ new String[] {
+ "1600-01-01 00:00:00.123456",
+ "1900-01-01 00:00:00.123456",
+ "1969-12-31 23:59:59.999999",
+ "1970-01-01 00:00:00",
+ "2020-07-17 18:00:22.123456",
+ }) {
+ assertThat(convert(defaultConverter, value))
+ .isEqualTo(convert(debeziumConverter, value));
+ }
+ }
+
+ @Test
+ void testPreEpochTimestampIsShiftedWhenDisabled() {
+ // Asia/Shanghai was UTC+08:05:43 before 1901, which makes the conversion of Debezium move
+ // the date and time fields of timestamps before 1970-01-01.
+ LocalDateTime converted =
+ (LocalDateTime) convert(defaultConverter, "1900-01-01 00:00:00.123456");
+ assertThat(converted).isNotEqualTo(LocalDateTime.parse("1900-01-01T00:00:00.123456"));
+ assertThat(convert(wallClockConverter, "1900-01-01 00:00:00.123456"))
+ .isEqualTo(LocalDateTime.parse("1900-01-01T00:00:00.123456"));
+ }
+
+ @Test
+ void testNullAndNonTimestampValues() {
+ for (CustomPostgresValueConverter converter :
+ new CustomPostgresValueConverter[] {defaultConverter, wallClockConverter}) {
+ assertThat(converter.convertTimestampToLocalDateTime(null, null, null)).isNull();
+
+ Object value = "1900-01-01 00:00:00";
+ assertThat(converter.convertTimestampToLocalDateTime(null, null, value))
+ .isSameAs(value);
+ }
+ }
+
+ @Test
+ void testInfinityTimestamps() {
+ for (CustomPostgresValueConverter converter :
+ new CustomPostgresValueConverter[] {defaultConverter, wallClockConverter}) {
+ assertThat(
+ converter.convertTimestampToLocalDateTime(
+ null, null, PostgresValueConverter.POSITIVE_INFINITY_TIMESTAMP))
+ .isEqualTo(PostgresValueConverter.POSITIVE_INFINITY_LOCAL_DATE_TIME);
+ assertThat(
+ converter.convertTimestampToLocalDateTime(
+ null, null, PostgresValueConverter.NEGATIVE_INFINITY_TIMESTAMP))
+ .isEqualTo(PostgresValueConverter.NEGATIVE_INFINITY_LOCAL_DATE_TIME);
+ }
+ }
+
+ private static Object convert(PostgresValueConverter converter, String value) {
+ return converter.convertTimestampToLocalDateTime(null, null, Timestamp.valueOf(value));
+ }
+
+ /**
+ * Tests that the option also reaches the value converter when it is passed as a Debezium
+ * property, which is how the DataStream API and the Postgres pipeline connector enable it.
+ */
+ @Test
+ void testOptionReadFromDebeziumPropertiesOfSourceConfig() {
+ Properties debeziumProperties = new Properties();
+ debeziumProperties.setProperty(
+ PostgresSourceOptions.SCAN_PRE_EPOCH_TIMESTAMP_WALL_CLOCK_CONVERSION_ENABLED.key(),
+ "true");
+
+ PostgresSourceConfigFactory configFactory = new PostgresSourceConfigFactory();
+ configFactory.hostname("localhost");
+ configFactory.port(5432);
+ configFactory.username("postgres");
+ configFactory.password("postgres");
+ configFactory.database("postgres");
+ configFactory.schemaList(new String[] {"inventory"});
+ configFactory.tableList("inventory.full_types");
+ configFactory.decodingPluginName("pgoutput");
+ configFactory.debeziumProperties(debeziumProperties);
+
+ PostgresSourceConfig sourceConfig = configFactory.create(0);
+
+ CustomPostgresValueConverter converter =
+ CustomPostgresValueConverter.of(
+ sourceConfig.getDbzConnectorConfig(), StandardCharsets.UTF_8, null);
+ assertThat(convert(converter, "1900-01-01 00:00:00.123"))
+ .isEqualTo(LocalDateTime.parse("1900-01-01T00:00:00.123"));
+ }
+
+ private static CustomPostgresValueConverter customConverter(boolean optionEnabled) {
+ Configuration configuration =
+ TestHelper.defaultConfig()
+ .with(
+ PostgresSourceOptions
+ .SCAN_PRE_EPOCH_TIMESTAMP_WALL_CLOCK_CONVERSION_ENABLED
+ .key(),
+ String.valueOf(optionEnabled))
+ .build();
+ return CustomPostgresValueConverter.of(
+ new PostgresConnectorConfig(configuration), StandardCharsets.UTF_8, null);
+ }
+
+ private static PostgresValueConverter debeziumConverter() {
+ return PostgresValueConverter.of(
+ new PostgresConnectorConfig(TestHelper.defaultConfig().build()),
+ StandardCharsets.UTF_8,
+ null);
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLConnectorITCase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLConnectorITCase.java
index c3f117076..527872360 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLConnectorITCase.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLConnectorITCase.java
@@ -45,11 +45,16 @@ import org.testcontainers.utility.DockerImageName;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
+import java.util.TimeZone;
import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
import java.util.stream.Stream;
import static org.testcontainers.containers.PostgreSQLContainer.POSTGRESQL_PORT;
@@ -713,6 +718,169 @@ class PostgreSQLConnectorITCase extends PostgresTestBase {
result.getJobClient().get().cancel().get();
}
+ @Test
+ void testPreEpochTimestampWallClockConversion() throws Throwable {
+ TimeZone originalTimeZone = TimeZone.getDefault();
+ try {
+ TimeZone.setDefault(TimeZone.getTimeZone("Asia/Shanghai"));
+ setup(true);
+ initializePostgresTable(POSTGIS_CONTAINER, "column_type_test");
+ insertPreEpochTimestampRow();
+
+ String sourceDDL = preEpochTimestampSourceDdl("true");
+ String sinkDDL =
+ "CREATE TABLE sink ("
+ + " id INTEGER NOT NULL,"
+ + " small_c SMALLINT,"
+ + " timestamp3_c TIMESTAMP(3),"
+ + " timestamp6_c TIMESTAMP(6),"
+ + " PRIMARY KEY (id) NOT ENFORCED"
+ + ") WITH ("
+ + " 'connector' = 'values',"
+ + " 'sink-insert-only' = 'false'"
+ + ")";
+ tEnv.executeSql(sourceDDL);
+ tEnv.executeSql(sinkDDL);
+
+ TableResult result = tEnv.executeSql("INSERT INTO sink SELECT * FROM full_types");
+
+ waitForSinkSize("sink", 2);
+ // the stored value is read as it is during the snapshot phase
+ Assertions.assertThat(preEpochSnapshotRecords())
+ .containsExactly(
+ "+I(2,32767,1900-01-01T00:00:00.123,1900-01-01T00:00:00.123456)");
+
+ // wait a bit to make sure the replication slot is ready
+ Thread.sleep(5000);
+ try (Connection connection = getJdbcConnection(POSTGIS_CONTAINER);
+ Statement statement = connection.createStatement()) {
+ statement.execute("UPDATE inventory.full_types SET small_c=0 WHERE id=2;");
+ }
+
+ waitForSinkSize("sink", 4);
+ // the stored value is also kept during the change log phase
+ List<String> expected =
+ Arrays.asList(
+ "+I(2,32767,1900-01-01T00:00:00.123,1900-01-01T00:00:00.123456)",
+ "-D(2,32767,1900-01-01T00:00:00.123,1900-01-01T00:00:00.123456)",
+ "+I(2,0,1900-01-01T00:00:00.123,1900-01-01T00:00:00.123456)");
+ Assertions.assertThat(preEpochChangeLogRecords()).containsExactlyElementsOf(expected);
+
+ result.getJobClient().get().cancel().get();
+ } finally {
+ TimeZone.setDefault(originalTimeZone);
+ }
+ }
+
+ @Test
+ void testPreEpochTimestampConversionUnchangedByDefault() throws Throwable {
+ TimeZone originalTimeZone = TimeZone.getDefault();
+ try {
+ TimeZone.setDefault(TimeZone.getTimeZone("Asia/Shanghai"));
+ setup(true);
+ initializePostgresTable(POSTGIS_CONTAINER, "column_type_test");
+ insertPreEpochTimestampRow();
+
+ // the option is not configured, so the previous conversion behavior is expected
+ String sourceDDL = preEpochTimestampSourceDdl(null);
+ String sinkDDL =
+ "CREATE TABLE sink ("
+ + " id INTEGER NOT NULL,"
+ + " small_c SMALLINT,"
+ + " timestamp3_c TIMESTAMP(3),"
+ + " timestamp6_c TIMESTAMP(6),"
+ + " PRIMARY KEY (id) NOT ENFORCED"
+ + ") WITH ("
+ + " 'connector' = 'values',"
+ + " 'sink-insert-only' = 'false'"
+ + ")";
+ tEnv.executeSql(sourceDDL);
+ tEnv.executeSql(sinkDDL);
+
+ TableResult result = tEnv.executeSql("INSERT INTO sink SELECT * FROM full_types");
+
+ waitForSinkSize("sink", 2);
+ String timestamp3 = legacyConvertedTimestamp("1900-01-01 00:00:00.123");
+ String timestamp6 = legacyConvertedTimestamp("1900-01-01 00:00:00.123456");
+ Assertions.assertThat(preEpochSnapshotRecords())
+ .containsExactly(String.format("+I(2,32767,%s,%s)", timestamp3, timestamp6));
+
+ result.getJobClient().get().cancel().get();
+ } finally {
+ TimeZone.setDefault(originalTimeZone);
+ }
+ }
+
+ private void insertPreEpochTimestampRow() throws SQLException {
+ try (Connection connection = getJdbcConnection(POSTGIS_CONTAINER);
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ "INSERT INTO inventory.full_types "
+ + "(id, small_c, timestamp3_c, timestamp6_c) VALUES "
+ + "(2, 32767, '1900-01-01 00:00:00.123', "
+ + "'1900-01-01 00:00:00.123456');");
+ }
+ }
+
+ private String preEpochTimestampSourceDdl(String wallClockConversionEnabled) {
+ String option =
+ wallClockConversionEnabled == null
+ ? ""
+ : " 'scan.pre-epoch-timestamp.wall-clock-conversion.enabled' = '"
+ + wallClockConversionEnabled
+ + "',";
+ return String.format(
+ "CREATE TABLE full_types ("
+ + " id INTEGER NOT NULL,"
+ + " small_c SMALLINT,"
+ + " timestamp3_c TIMESTAMP(3),"
+ + " timestamp6_c TIMESTAMP(6)"
+ + ") WITH ("
+ + " 'connector' = 'postgres-cdc',"
+ + " 'hostname' = '%s',"
+ + " 'port' = '%s',"
+ + " 'username' = '%s',"
+ + " 'password' = '%s',"
+ + " 'database-name' = '%s',"
+ + " 'schema-name' = '%s',"
+ + " 'table-name' = '%s',"
+ + " 'scan.incremental.snapshot.enabled' = 'true',"
+ + option
+ + " 'decoding.plugin.name' = 'pgoutput', "
+ + " 'slot.name' = '%s'"
+ + ")",
+ POSTGIS_CONTAINER.getHost(),
+ POSTGIS_CONTAINER.getMappedPort(POSTGRESQL_PORT),
+ POSTGIS_CONTAINER.getUsername(),
+ POSTGIS_CONTAINER.getPassword(),
+ POSTGIS_CONTAINER.getDatabaseName(),
+ "inventory",
+ "full_types",
+ getSlotName());
+ }
+
+ private List<String> preEpochSnapshotRecords() {
+ return TestValuesTableFactory.getRawResultsAsStrings("sink").stream()
+ .filter(record -> record.startsWith("+I(2,"))
+ .collect(Collectors.toList());
+ }
+
+ private List<String> preEpochChangeLogRecords() {
+ return TestValuesTableFactory.getRawResultsAsStrings("sink").stream()
+ .filter(record -> record.startsWith("+I(2,") || record.startsWith("-D(2,"))
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Returns the string of a stored {@code timestamp} value before 1970-01-01 after the conversion
+ * which Debezium does, i.e. through an instant with the default time zone of the JVM.
+ */
+ private static String legacyConvertedTimestamp(String timestamp) {
+ return LocalDateTime.ofInstant(
+ Timestamp.valueOf(timestamp).toInstant(), ZoneId.systemDefault())
+ .toString();
+ }
+
@ParameterizedTest
@ValueSource(booleans = {true, false})
void testMetadataColumns(boolean parallelismSnapshot) throws Throwable {