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 <[email protected]>
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 {
