lvyanquan commented on code in PR #4214:
URL: https://github.com/apache/flink-cdc/pull/4214#discussion_r4080036561
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLConnectorITCase.java:
##########
@@ -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);
Review Comment:
Thanks. I restored `testAllTypes` to its original ordered assertion and
moved the pre-epoch regression coverage into a dedicated
`testPreEpochTimestamp` ITCase. The new test filters the target row but
verifies the exact `+I -> -D -> +I` sequence, so event ordering remains covered
without relying on cross-row snapshot ordering.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/CustomPostgresValueConverter.java:
##########
@@ -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());
+ }
+
+ @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();
Review Comment:
Thanks. I added method-level Javadoc explaining that `column` and
`fieldDefn` are intentionally unused because this override only changes how the
JDBC `Timestamp` value is converted. Schema and field-specific handling remains
in the parent converter flow.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/CustomPostgresValueConverter.java:
##########
@@ -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());
+ }
+
Review Comment:
Addressed. The class and overridden method now document why the
instant/system-default-zone round trip is incorrect for pre-epoch timestamps in
regional time zones, why `Timestamp.toLocalDateTime()` preserves the wall-clock
fields, and that the change applies only to PostgreSQL `timestamp` without time
zone.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/CustomPostgresValueConverter.java:
##########
@@ -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());
+ }
+
+ @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();
+ }
Review Comment:
Addressed. I added `CustomPostgresValueConverterTest` covering null and
non-`Timestamp` values, positive and negative infinity sentinels, timestamps in
1600 and 1900, immediately before and at the 1970 epoch boundary, and a
post-epoch value under `Asia/Shanghai`. The test passes with both Flink 1.20
and Flink 2.2.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
