[FLINK-38742][cdc/postgres] Add support for TIMESTAMP WITH TIME ZONE in Debezium event deserialization

Author: Fluder-Paradyne

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSchemaDataTypeInference.java

@@ -25,14 +25,46 @@
 import io.debezium.data.geometry.Geography;
 import io.debezium.data.geometry.Geometry;
 import io.debezium.data.geometry.Point;
+import io.debezium.time.ZonedTimestamp;
 import org.apache.kafka.connect.data.Schema;
 
+import java.time.Instant;
+import java.util.Optional;
+
 /** {@link DataType} inference for PostgresSQL debezium {@link Schema}. */
 @Internal
 public class PostgresSchemaDataTypeInference extends DebeziumSchemaDataTypeInference {
 
     private static final long serialVersionUID = 1L;
 
+    protected DataType inferString(Object value, Schema schema) {
+        // PostgreSQL TIMESTAMPTZ is encoded as ZonedTimestamp in Debezium
+        // We need to return TIMESTAMP_TZ (ZonedTimestampType) instead of TIMESTAMP_LTZ
+        if (ZonedTimestamp.SCHEMA_NAME.equals(schema.name())) {
+            int nano =
+                    Optional.ofNullable((String) value)
+                            .map(s -> ZonedTimestamp.FORMATTER.parse(s, Instant::from))
+                            .map(Instant::getNano)
+                            .orElse(0);
+
+            int precision;
+            if (nano == 0) {
+                precision = 0;
+            } else if (nano % 1000 > 0) {
+                precision = 9;
+            } else if (nano % 1000_000 > 0) {
+                precision = 6;
+            } else if (nano % 1000_000_000 > 0) {
+                precision = 3;
+            } else {
+                precision = 0;
+            }
+            // Return TIMESTAMP_TZ (ZonedTimestampType) for PostgreSQL TIMESTAMPTZ
+            return DataTypes.TIMESTAMP_TZ(precision);
+        }
+        return super.inferString(value, schema);
+    }
+
     protected DataType inferStruct(Object value, Schema schema) {
         // the Geometry datatype in PostgresSQL will be converted to
         // a String with Json format

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java

@@ -27,6 +27,7 @@
 import org.apache.flink.cdc.common.data.RecordData;
 import org.apache.flink.cdc.common.data.TimeData;
 import org.apache.flink.cdc.common.data.TimestampData;
+import org.apache.flink.cdc.common.data.ZonedTimestampData;
 import org.apache.flink.cdc.common.data.binary.BinaryStringData;
 import org.apache.flink.cdc.common.event.ChangeEvent;
 import org.apache.flink.cdc.common.event.CreateTableEvent;
@@ -54,6 +55,7 @@
 import io.debezium.time.NanoTime;
 import io.debezium.time.NanoTimestamp;
 import io.debezium.time.Timestamp;
+import io.debezium.time.ZonedTimestamp;
 import org.apache.kafka.connect.data.Decimal;
 import org.apache.kafka.connect.data.Field;
 import org.apache.kafka.connect.data.Schema;
@@ -65,6 +67,7 @@
 import java.math.BigDecimal;
 import java.nio.ByteBuffer;
 import java.time.Instant;
+import java.time.OffsetDateTime;
 import java.util.Collections;
 import java.util.Date;
 import java.util.HashMap;
@@ -196,6 +199,8 @@ protected DeserializationRuntimeConverter createNotNullConverter(DataType type)
                 return this::convertToTime;
             case TIMESTAMP_WITHOUT_TIME_ZONE:
                 return this::convertToTimestamp;
+            case TIMESTAMP_WITH_TIME_ZONE:
+                return this::convertToZonedTimestamp;
             case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
                 return this::convertToLocalTimeZoneTimestamp;
             case FLOAT:
@@ -367,6 +372,28 @@ protected Object convertToTimestamp(Object dbzObj, Schema schema) {
                         + dbzObj.getClass().getName());
     }
 
+    protected Object convertToZonedTimestamp(Object dbzObj, Schema schema) {
+        if (dbzObj instanceof String) {
+            String str = (String) dbzObj;
+            // ZonedTimestamp type is encoded in string type with timezone offset
+            // Format: ISO-8601 with timezone offset (e.g., "2020-07-17T18:00:22+00:00")
+            if (ZonedTimestamp.SCHEMA_NAME.equals(schema.name())) {
+                // Parse using Debezium's ZonedTimestamp formatter
+                OffsetDateTime offsetDateTime = OffsetDateTime.parse(str, ZonedTimestamp.FORMATTER);
+                return ZonedTimestampData.fromOffsetDateTime(offsetDateTime);
+            } else {
+                // Fallback to standard ISO-8601 parsing
+                OffsetDateTime offsetDateTime = OffsetDateTime.parse(str);
+                return ZonedTimestampData.fromOffsetDateTime(offsetDateTime);
+            }
+        }
+        throw new IllegalArgumentException(
+                "Unable to convert to TIMESTAMP WITH TIME ZONE from unexpected value '"
+                        + dbzObj
+                        + "' of type "
+                        + dbzObj.getClass().getName());
+    }
+
     protected Object convertToLocalTimeZoneTimestamp(Object dbzObj, Schema schema) {
         if (dbzObj instanceof String) {
             String str = (String) dbzObj;