[cdc] Add more checker for the timestamp type. (#3312)

apache · May 13, 2024 · 3a7d030 · 3a7d030
1 parent fa43a67
commit 3a7d030
Show file tree

Hide file tree

Showing 2 changed files with 30 additions and 0 deletions.
diff --git a/.../src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java b/.../src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java
@@ -63,6 +63,9 @@ public abstract class UpdatedDataFieldsProcessFunctionBase<I, O> extends Process
 
     private static final List<DataTypeRoot> DECIMAL_TYPES = Arrays.asList(DataTypeRoot.DECIMAL);
 
+    private static final List<DataTypeRoot> TIMESTAMP_TYPES =
+            Arrays.asList(DataTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE);
+
     protected UpdatedDataFieldsProcessFunctionBase(Catalog.Loader catalogLoader) {
         this.catalogLoader = catalogLoader;
     }
@@ -174,6 +177,14 @@ public static ConvertAction canConvert(DataType oldType, DataType newType) {
                     : ConvertAction.CONVERT;
         }
 
+        oldIdx = TIMESTAMP_TYPES.indexOf(oldType.getTypeRoot());
+        newIdx = TIMESTAMP_TYPES.indexOf(newType.getTypeRoot());
+        if (oldIdx >= 0 && newIdx >= 0) {
+            return DataTypeChecks.getPrecision(oldType) <= DataTypeChecks.getPrecision(newType)
+                    ? ConvertAction.CONVERT
+                    : ConvertAction.IGNORE;
+        }
+
         return ConvertAction.EXCEPTION;
     }
 

diff --git a/.../test/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBaseTest.java b/.../test/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBaseTest.java
@@ -22,6 +22,7 @@
 import org.apache.paimon.types.DecimalType;
 import org.apache.paimon.types.IntType;
 import org.apache.paimon.types.SmallIntType;
+import org.apache.paimon.types.TimestampType;
 import org.apache.paimon.types.VarCharType;
 
 import org.junit.Assert;
@@ -79,4 +80,22 @@ public void testCanConvertDecimal() {
         Assert.assertEquals(
                 UpdatedDataFieldsProcessFunctionBase.ConvertAction.IGNORE, convertAction);
     }
+
+    @Test
+    public void testCanConvertTimestamp() {
+        TimestampType oldType = new TimestampType(true, 3);
+        TimestampType biggerLengthTimestamp = new TimestampType(true, 5);
+        TimestampType smallerLengthTimestamp = new TimestampType(true, 2);
+
+        UpdatedDataFieldsProcessFunctionBase.ConvertAction convertAction = null;
+        convertAction =
+                UpdatedDataFieldsProcessFunctionBase.canConvert(oldType, biggerLengthTimestamp);
+        Assert.assertEquals(
+                UpdatedDataFieldsProcessFunctionBase.ConvertAction.CONVERT, convertAction);
+        convertAction =
+                UpdatedDataFieldsProcessFunctionBase.canConvert(oldType, smallerLengthTimestamp);
+
+        Assert.assertEquals(
+                UpdatedDataFieldsProcessFunctionBase.ConvertAction.IGNORE, convertAction);
+    }
 }