This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 1f0121906e [cdc] Fix PostgreSQL time(n) mapping and widen TIME
precision in schema evolution (#8609)
1f0121906e is described below
commit 1f0121906e67610b84361590fe0566bd544b32f4
Author: Eunbin Son <[email protected]>
AuthorDate: Thu Jul 23 22:14:47 2026 +0900
[cdc] Fix PostgreSQL time(n) mapping and widen TIME precision in schema
evolution (#8609)
---
.../action/cdc/postgres/PostgresRecordParser.java | 3 ++
.../cdc/UpdatedDataFieldsProcessFunctionBase.java | 14 ++++++
.../cdc/postgres/PostgresRecordParserTest.java | 51 +++++++++++++++++++++-
.../UpdatedDataFieldsProcessFunctionBaseTest.java | 19 ++++++++
4 files changed, 86 insertions(+), 1 deletion(-)
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresRecordParser.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresRecordParser.java
index 9375c1630e..003ae7b529 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresRecordParser.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresRecordParser.java
@@ -44,6 +44,7 @@ import io.debezium.data.Bits;
import io.debezium.time.Date;
import io.debezium.time.MicroTime;
import io.debezium.time.MicroTimestamp;
+import io.debezium.time.Time;
import io.debezium.time.Timestamp;
import io.debezium.time.ZonedTimestamp;
import org.apache.flink.api.common.functions.FlatMapFunction;
@@ -159,6 +160,8 @@ public class PostgresRecordParser
case "int32":
if (Date.SCHEMA_NAME.equals(field.name())) {
return DataTypes.DATE();
+ } else if (Time.SCHEMA_NAME.equals(field.name())) {
+ return DataTypes.TIME(3);
}
return DataTypes.INT();
case "int64":
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java
index 6f74a346e9..877fc8d20b 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java
@@ -82,6 +82,9 @@ public abstract class UpdatedDataFieldsProcessFunctionBase<I,
O> extends Process
private static final List<DataTypeRoot> TIMESTAMP_TYPES =
Arrays.asList(DataTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE);
+ private static final List<DataTypeRoot> TIME_TYPES =
+ Arrays.asList(DataTypeRoot.TIME_WITHOUT_TIME_ZONE);
+
protected UpdatedDataFieldsProcessFunctionBase(
CatalogLoader catalogLoader, TypeMapping typeMapping) {
this.catalogLoader = catalogLoader;
@@ -260,6 +263,17 @@ public abstract class
UpdatedDataFieldsProcessFunctionBase<I, O> extends Process
: ConvertAction.IGNORE;
}
+ oldIdx = TIME_TYPES.indexOf(oldType.getTypeRoot());
+ newIdx = TIME_TYPES.indexOf(newType.getTypeRoot());
+ if (oldIdx >= 0 && newIdx >= 0) {
+ // Debezium encodes every PostgreSQL time(0..3) column as millis
(TIME(3)), while the
+ // JDBC schema path keeps the declared TIME(n); widen on precision
so time(0..2) can
+ // converge on TIME(3) instead of failing schema evolution.
+ return DataTypeChecks.getPrecision(oldType) <=
DataTypeChecks.getPrecision(newType)
+ ? ConvertAction.CONVERT
+ : ConvertAction.IGNORE;
+ }
+
return ConvertAction.EXCEPTION;
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresRecordParserTest.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresRecordParserTest.java
index f75fcdf61e..5db746f74d 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresRecordParserTest.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/postgres/PostgresRecordParserTest.java
@@ -88,6 +88,44 @@ public class PostgresRecordParserTest {
.isEqualTo(DataTypes.BIGINT().nullable());
}
+ /**
+ * Verifies that {@code io.debezium.time.Time} (int32,
milliseconds-past-midnight) is mapped to
+ * {@code TIME(3)}, not {@code INT}.
+ *
+ * <p>PostgreSQL {@code time(n)} columns with {@code n <= 3} are encoded
by Debezium (default
+ * adaptive mode) using the {@code io.debezium.time.Time} logical type
(int32 millis-of-day).
+ * The JDBC schema path maps the same columns to {@code TIME(n)}, so
without this fix the two
+ * paths disagree and the sync crashes with "Cannot convert field from
TIME(3) to INT". This
+ * mirrors the int64 {@code Timestamp} fix from PR #8222.
+ */
+ @Test
+ public void testTimeMillisFieldMapsToTime3() throws Exception {
+ String json = debeziumJson("int32", "io.debezium.time.Time");
+ List<RichCdcMultiplexRecord> out = parse(json);
+
+ assertThat(out).isNotEmpty();
+ DataField field = findField(out.get(0), "ts_col");
+ assertThat(field.type())
+ .as("io.debezium.time.Time (int32) must map to TIME(3), not
INT")
+ .isEqualTo(DataTypes.TIME(3).nullable());
+ }
+
+ /**
+ * Verifies that a plain int32 field (no logical type name) still maps to
INT — i.e. the fix
+ * does not break the default fallthrough.
+ */
+ @Test
+ public void testPlainInt32FieldMapsToInt() throws Exception {
+ String json = debeziumJson("int32", null);
+ List<RichCdcMultiplexRecord> out = parse(json);
+
+ assertThat(out).isNotEmpty();
+ DataField field = findField(out.get(0), "ts_col");
+ assertThat(field.type())
+ .as("int32 with no logical type name must remain INT")
+ .isEqualTo(DataTypes.INT().nullable());
+ }
+
//
-------------------------------------------------------------------------
// helpers
//
-------------------------------------------------------------------------
@@ -122,6 +160,15 @@ public class PostgresRecordParserTest {
* ts_col} with the given logical type {@code schemaName} (may be null for
a plain int64).
*/
private static String debeziumJson(String schemaName) {
+ return debeziumJson("int64", schemaName);
+ }
+
+ /**
+ * Builds a minimal Debezium PostgreSQL CDC JSON event containing one
column {@code ts_col} of
+ * the given primitive {@code type} and logical type {@code schemaName}
(may be null for no
+ * logical type).
+ */
+ private static String debeziumJson(String type, String schemaName) {
String nameField = schemaName == null ? "" : "\"name\":\"" +
schemaName + "\",";
return "{"
+ "\"schema\":{"
@@ -133,7 +180,9 @@ public class PostgresRecordParserTest {
+ " \"field\":\"after\","
+ " \"fields\":["
+ "
{\"type\":\"int32\",\"optional\":false,\"field\":\"id\"},"
- + " {\"type\":\"int64\",\"optional\":true,"
+ + " {\"type\":\""
+ + type
+ + "\",\"optional\":true,"
+ nameField
+ " \"field\":\"ts_col\"}"
+ " ]"
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBaseTest.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBaseTest.java
index a201e8308c..0307cee204 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBaseTest.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBaseTest.java
@@ -29,6 +29,7 @@ import org.apache.paimon.types.MapType;
import org.apache.paimon.types.MultisetType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.types.SmallIntType;
+import org.apache.paimon.types.TimeType;
import org.apache.paimon.types.TimestampType;
import org.apache.paimon.types.VarCharType;
@@ -95,6 +96,24 @@ public class UpdatedDataFieldsProcessFunctionBaseTest {
assertEquals(UpdatedDataFieldsProcessFunctionBase.ConvertAction.IGNORE,
convertAction);
}
+ @Test
+ public void testCanConvertTime() {
+ // Debezium maps every PostgreSQL time(0..3) column to TIME(3), while
the JDBC schema path
+ // keeps TIME(n); widening TIME(0) -> TIME(3) must convert instead of
throwing.
+ TimeType oldType = new TimeType(true, 0);
+ TimeType biggerPrecision = new TimeType(true, 3);
+ TimeType smallerPrecision = new TimeType(true, 0);
+
+ assertEquals(
+ UpdatedDataFieldsProcessFunctionBase.ConvertAction.CONVERT,
+ UpdatedDataFieldsProcessFunctionBase.canConvert(
+ oldType, biggerPrecision,
TypeMapping.defaultMapping()));
+ assertEquals(
+ UpdatedDataFieldsProcessFunctionBase.ConvertAction.IGNORE,
+ UpdatedDataFieldsProcessFunctionBase.canConvert(
+ biggerPrecision, smallerPrecision,
TypeMapping.defaultMapping()));
+ }
+
@Test
public void testCanConvertTimestamp() {
TimestampType oldType = new TimestampType(true, 3);
