Skip to content

[FLINK-38835][postgres] Fix timestamp conversion to LocalDateTime for dates before 1970-01-01. #4214

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lvyanquan wants to merge 1 commit into
base: master
Choose a base branch
from
+116 −6
Open

[FLINK-38835][postgres] Fix timestamp conversion to LocalDateTime for dates before 1970-01-01. #4214

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
99 changes: 99 additions & 0 deletions ...gres-cdc/src/main/java/io/debezium/connector/postgresql/CustomPostgresValueConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,99 @@
/*
* 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 io.debezium.config.CommonConnectorConfig;
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 custom PostgresValueConverter that correctly handles timestamp conversion to LocalDateTime for
* dates before 1970-01-01.
*/
public class CustomPostgresValueConverter extends PostgresValueConverter {
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) {
super(
databaseCharset,
decimalMode,
temporalPrecisionMode,
defaultOffset,
bigIntUnsignedMode,
includeUnknownDatatypes,
typeRegistry,
hStoreMode,
binaryMode,
intervalMode,
toastPlaceholder,
moneyFractionDigits);
}

public static CustomPostgresValueConverter of(
PostgresConnectorConfig connectorConfig,
Charset databaseCharset,
TypeRegistry typeRegistry) {
return new CustomPostgresValueConverter(
databaseCharset,
connectorConfig.getDecimalMode(),
connectorConfig.getTemporalPrecisionMode(),
ZoneOffset.UTC,
null,
connectorConfig.includeUnknownDatatypes(),
typeRegistry,
connectorConfig.hStoreHandlingMode(),
connectorConfig.binaryHandlingMode(),
connectorConfig.intervalHandlingMode(),
connectorConfig.getUnavailableValuePlaceholder(),
connectorConfig.moneyFractionDigits());
}

Copy link

Copilot AI Jan 16, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The overridden method convertTimestampToLocalDateTime should include a JavaDoc comment explaining why this override is necessary and how it differs from the parent implementation. Specifically, it should document that this method uses Timestamp.toLocalDateTime() to properly handle dates before 1970-01-01 by avoiding timezone offset inconsistencies.

Suggested change
/**
* Overrides {@link PostgresValueConverter#convertTimestampToLocalDateTime(Column, Field, Object)}
* to use {@link Timestamp#toLocalDateTime()} directly.
*
* <p>The default implementation derives the {@code LocalDateTime} value using epoch-based
* calculations, which can apply timezone offsets inconsistently for timestamps before
* {@code 1970-01-01}. By delegating to {@code Timestamp.toLocalDateTime()}, this override
* avoids those pre-epoch timezone offset inconsistencies while preserving the behavior for
* timestamps on or after the epoch.</p>
*/

Copilot uses AI. Check for mistakes.
@Override
protected Object convertTimestampToLocalDateTime(Column column, Field fieldDefn, Object 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;
}

return timestamp.toLocalDateTime();
Comment on lines +82 to +97
Copy link

Copilot AI Jan 16, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The overridden method doesn't use the column and fieldDefn parameters. If the parent class implementation uses these parameters for specific logic (such as handling different column types or field definitions), removing this logic could cause unexpected behavior. Consider adding a comment explaining why these parameters can be safely ignored in this implementation.

Copilot uses AI. Check for mistakes.
}
Comment on lines +81 to +98
Copy link

Copilot AI Jan 16, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The CustomPostgresValueConverter class lacks dedicated unit tests. While integration tests are added, unit tests should verify the convertTimestampToLocalDateTime method's behavior with various edge cases including null values, non-Timestamp objects, infinity timestamps, and crucially, timestamps both before and after 1970-01-01 to ensure the timezone handling fix works correctly.

Copilot uses AI. Check for mistakes.
}
2 changes: 1 addition & 1 deletion ...ctor-postgres-cdc/src/main/java/io/debezium/connector/postgresql/PostgresObjectUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,7 @@ public static PostgresEventMetadataProvider newEventMetadataProvider() {
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
Expand Down
4 changes: 3 additions & 1 deletion ...res-cdc/src/main/java/io/debezium/connector/postgresql/connection/PostgresConnection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
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;
Expand Down Expand Up @@ -188,7 +189,8 @@ public PostgresConnection(
} 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());
}
Expand Down
11 changes: 7 additions & 4 deletions ...c/test/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLConnectorITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -689,18 +689,21 @@ void testAllTypes(boolean parallelismSnapshot) throws Throwable {
// generate WAL
try (Connection connection = getJdbcConnection(POSTGIS_CONTAINER);
Statement statement = connection.createStatement()) {
statement.execute("UPDATE inventory.full_types SET small_c=0 WHERE id=1;");
statement.execute("UPDATE inventory.full_types SET small_c=0 WHERE id<=2;");
}

waitForSinkSize("sink", 3);
waitForSinkSize("sink", 6);

List<String> expected =
Arrays.asList(
"+I(1,[50],32767,65535,2147483647,5.5,6.6,123.12345,404.4,true,Hello World,a,abc,abcd..xyz,2020-07-17T18:00:22.123,2020-07-17T18:00:22.123456,2020-07-17,18:00:22,500,{\"hexewkb\":\"0105000020e610000001000000010200000002000000a779c7293a2465400b462575025a46c0c66d3480b7fc6440c3d32b65195246c0\",\"srid\":4326},{\"hexewkb\":\"0101000020730c00001c7c613255de6540787aa52c435c42c0\",\"srid\":3187})",
"+I(2,[50],32767,65535,2147483647,5.5,6.6,123.12345,404.4,true,Hello World,a,abc,abcd..xyz,1900-01-01T00:00:00.123,1900-01-01T00:00:00.123456,1900-01-01,18:00:22,500,{\"hexewkb\":\"0105000020e610000001000000010200000002000000a779c7293a2465400b462575025a46c0c66d3480b7fc6440c3d32b65195246c0\",\"srid\":4326},{\"hexewkb\":\"0101000020730c00001c7c613255de6540787aa52c435c42c0\",\"srid\":3187})",
"-D(1,[50],32767,65535,2147483647,5.5,6.6,123.12345,404.4,true,Hello World,a,abc,abcd..xyz,2020-07-17T18:00:22.123,2020-07-17T18:00:22.123456,2020-07-17,18:00:22,500,{\"hexewkb\":\"0105000020e610000001000000010200000002000000a779c7293a2465400b462575025a46c0c66d3480b7fc6440c3d32b65195246c0\",\"srid\":4326},{\"hexewkb\":\"0101000020730c00001c7c613255de6540787aa52c435c42c0\",\"srid\":3187})",
"+I(1,[50],0,65535,2147483647,5.5,6.6,123.12345,404.4,true,Hello World,a,abc,abcd..xyz,2020-07-17T18:00:22.123,2020-07-17T18:00:22.123456,2020-07-17,18:00:22,500,{\"hexewkb\":\"0105000020e610000001000000010200000002000000a779c7293a2465400b462575025a46c0c66d3480b7fc6440c3d32b65195246c0\",\"srid\":4326},{\"hexewkb\":\"0101000020730c00001c7c613255de6540787aa52c435c42c0\",\"srid\":3187})");
"-D(2,[50],32767,65535,2147483647,5.5,6.6,123.12345,404.4,true,Hello World,a,abc,abcd..xyz,1900-01-01T00:00:00.123,1900-01-01T00:00:00.123456,1900-01-01,18:00:22,500,{\"hexewkb\":\"0105000020e610000001000000010200000002000000a779c7293a2465400b462575025a46c0c66d3480b7fc6440c3d32b65195246c0\",\"srid\":4326},{\"hexewkb\":\"0101000020730c00001c7c613255de6540787aa52c435c42c0\",\"srid\":3187})",
"+I(1,[50],0,65535,2147483647,5.5,6.6,123.12345,404.4,true,Hello World,a,abc,abcd..xyz,2020-07-17T18:00:22.123,2020-07-17T18:00:22.123456,2020-07-17,18:00:22,500,{\"hexewkb\":\"0105000020e610000001000000010200000002000000a779c7293a2465400b462575025a46c0c66d3480b7fc6440c3d32b65195246c0\",\"srid\":4326},{\"hexewkb\":\"0101000020730c00001c7c613255de6540787aa52c435c42c0\",\"srid\":3187})",
"+I(2,[50],0,65535,2147483647,5.5,6.6,123.12345,404.4,true,Hello World,a,abc,abcd..xyz,1900-01-01T00:00:00.123,1900-01-01T00:00:00.123456,1900-01-01,18:00:22,500,{\"hexewkb\":\"0105000020e610000001000000010200000002000000a779c7293a2465400b462575025a46c0c66d3480b7fc6440c3d32b65195246c0\",\"srid\":4326},{\"hexewkb\":\"0101000020730c00001c7c613255de6540787aa52c435c42c0\",\"srid\":3187})");
List<String> actual = TestValuesTableFactory.getRawResultsAsStrings("sink");
Assertions.assertThat(actual).isEqualTo(expected);
Assertions.assertThat(actual).containsExactlyInAnyOrderElementsOf(expected);
Copy link

Copilot AI Jan 16, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The change from isEqualTo to containsExactlyInAnyOrderElementsOf weakens the test by no longer verifying event ordering. While this is appropriate for parallel processing scenarios (as indicated by the parallelismSnapshot parameter), consider documenting why order verification was removed or adding a separate test that verifies correct ordering when parallelismSnapshot is false.

Copilot uses AI. Check for mistakes.

result.getJobClient().get().cancel().get();
}
Expand Down
6 changes: 6 additions & 0 deletions ...ource-connectors/flink-connector-postgres-cdc/src/test/resources/ddl/column_type_test.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,4 +56,10 @@ INSERT INTO inventory.full_types
VALUES (1, '2', 32767, 65535, 2147483647, 5.5, 6.6, 123.12345, 404.4443, true,
'Hello World', 'a', 'abc', 'abcd..xyz', '2020-07-17 18:00:22.123', '2020-07-17 18:00:22.123456',
'2020-07-17', '18:00:22', 500, 'SRID=3187;POINT(174.9479 -36.7208)'::geometry,
'MULTILINESTRING((169.1321 -44.7032, 167.8974 -44.6414))'::geography);

INSERT INTO inventory.full_types
VALUES (2, '2', 32767, 65535, 2147483647, 5.5, 6.6, 123.12345, 404.4443, true,
'Hello World', 'a', 'abc', 'abcd..xyz', '1900-01-01 00:00:00.123', '1900-01-01 00:00:00.123456',
'1900-01-01', '18:00:22', 500, 'SRID=3187;POINT(174.9479 -36.7208)'::geometry,
'MULTILINESTRING((169.1321 -44.7032, 167.8974 -44.6414))'::geography);
Loading