You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by le...@apache.org on 2024/06/06 11:49:15 UTC
(flink-cdc) branch master updated: [FLINK-34908][pipeline-connector][doris] Fix MySQL to doris pipeline will lose precision for timestamp type
This is an automated email from the ASF dual-hosted git repository.
leonard pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new e2ccc836a [FLINK-34908][pipeline-connector][doris] Fix MySQL to doris pipeline will lose precision for timestamp type
e2ccc836a is described below
commit e2ccc836a056c16974e4956190bdce249705b7ee
Author: Xin Gong <ge...@gmail.com>
AuthorDate: Thu Jun 6 19:49:11 2024 +0800
[FLINK-34908][pipeline-connector][doris] Fix MySQL to doris pipeline will lose precision for timestamp type
This closes #3207.
---
.../doris/sink/DorisEventSerializer.java | 2 +-
.../doris/sink/DorisRowConverterTest.java | 39 +++++++++++++++++++---
2 files changed, 35 insertions(+), 6 deletions(-)
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-doris/src/main/java/org/apache/flink/cdc/connectors/doris/sink/DorisEventSerializer.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-doris/src/main/java/org/apache/flink/cdc/connectors/doris/sink/DorisEventSerializer.java
index a3d6f21d0..a1cb9182d 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-doris/src/main/java/org/apache/flink/cdc/connectors/doris/sink/DorisEventSerializer.java
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-doris/src/main/java/org/apache/flink/cdc/connectors/doris/sink/DorisEventSerializer.java
@@ -56,7 +56,7 @@ public class DorisEventSerializer implements DorisRecordSerializer<Event> {
/** Format timestamp-related type data. */
public static final DateTimeFormatter DATE_TIME_FORMATTER =
- DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+ DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSSSS");
/** ZoneId from pipeline config to support timestamp with local time zone. */
public final ZoneId pipelineZoneId;
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-doris/src/test/java/org/apache/flink/cdc/connectors/doris/sink/DorisRowConverterTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-doris/src/test/java/org/apache/flink/cdc/connectors/doris/sink/DorisRowConverterTest.java
index 92ca83119..5f47fe509 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-doris/src/test/java/org/apache/flink/cdc/connectors/doris/sink/DorisRowConverterTest.java
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-doris/src/test/java/org/apache/flink/cdc/connectors/doris/sink/DorisRowConverterTest.java
@@ -17,6 +17,7 @@
package org.apache.flink.cdc.connectors.doris.sink;
+import org.apache.flink.cdc.common.data.LocalZonedTimestampData;
import org.apache.flink.cdc.common.data.TimestampData;
import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
import org.apache.flink.cdc.common.data.binary.BinaryStringData;
@@ -29,6 +30,7 @@ import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
import org.junit.Assert;
import org.junit.Test;
+import java.time.Instant;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
@@ -54,13 +56,32 @@ public class DorisRowConverterTest {
Column.physicalColumn("f12", DataTypes.TIMESTAMP()),
Column.physicalColumn("f14", DataTypes.DATE()),
Column.physicalColumn("f15", DataTypes.CHAR(1)),
- Column.physicalColumn("f16", DataTypes.VARCHAR(256)));
+ Column.physicalColumn("f16", DataTypes.VARCHAR(256)),
+ Column.physicalColumn("f17", DataTypes.TIMESTAMP()),
+ Column.physicalColumn("f18", DataTypes.TIMESTAMP()),
+ Column.physicalColumn("f19", DataTypes.TIMESTAMP()),
+ Column.physicalColumn("f20", DataTypes.TIMESTAMP_LTZ()),
+ Column.physicalColumn("f21", DataTypes.TIMESTAMP_LTZ()),
+ Column.physicalColumn("f22", DataTypes.TIMESTAMP_LTZ()));
List<DataType> dataTypes =
columns.stream().map(v -> v.getType()).collect(Collectors.toList());
- LocalDateTime time1 = LocalDateTime.of(2021, 1, 1, 8, 0, 0);
+ LocalDateTime time1 =
+ LocalDateTime.ofInstant(Instant.parse("2021-01-01T08:00:00Z"), ZoneId.of("Z"));
LocalDate date1 = LocalDate.of(2021, 1, 1);
+ LocalDateTime f17 =
+ LocalDateTime.ofInstant(Instant.parse("2021-01-01T08:01:11Z"), ZoneId.of("Z"));
+ LocalDateTime f18 =
+ LocalDateTime.ofInstant(Instant.parse("2021-01-01T08:01:11.123Z"), ZoneId.of("Z"));
+ LocalDateTime f19 =
+ LocalDateTime.ofInstant(
+ Instant.parse("2021-01-01T08:01:11.123456Z"), ZoneId.of("Z"));
+
+ Instant f20 = Instant.parse("2021-01-01T08:01:11Z");
+ Instant f21 = Instant.parse("2021-01-01T08:01:11.123Z");
+ Instant f22 = Instant.parse("2021-01-01T08:01:11.123456Z");
+
BinaryRecordDataGenerator generator =
new BinaryRecordDataGenerator(RowType.of(dataTypes.toArray(new DataType[] {})));
BinaryRecordData recordData =
@@ -76,17 +97,25 @@ public class DorisRowConverterTest {
TimestampData.fromLocalDateTime(time1),
(int) date1.toEpochDay(),
BinaryStringData.fromString("a"),
- BinaryStringData.fromString("doris")
+ BinaryStringData.fromString("doris"),
+ TimestampData.fromLocalDateTime(f17),
+ TimestampData.fromLocalDateTime(f18),
+ TimestampData.fromLocalDateTime(f19),
+ LocalZonedTimestampData.fromInstant(f20),
+ LocalZonedTimestampData.fromInstant(f21),
+ LocalZonedTimestampData.fromInstant(f22),
});
List row = new ArrayList();
for (int i = 0; i < recordData.getArity(); i++) {
DorisRowConverter.SerializationConverter converter =
DorisRowConverter.createNullableExternalConverter(
- columns.get(i).getType(), ZoneId.systemDefault());
+ columns.get(i).getType(), ZoneId.of("GMT+08:00"));
row.add(converter.serialize(i, recordData));
}
Assert.assertEquals(
- "[true, 1.2, 1.2345, 1, 32, 64, 128, 2021-01-01 08:00:00, 2021-01-01, a, doris]",
+ "[true, 1.2, 1.2345, 1, 32, 64, 128, 2021-01-01 08:00:00.000000, 2021-01-01, a, doris, 2021-01-01 "
+ + "08:01:11.000000, 2021-01-01 08:01:11.123000, 2021-01-01 08:01:11.123456, 2021-01-01 "
+ + "16:01:11.000000, 2021-01-01 16:01:11.123000, 2021-01-01 16:01:11.123456]",
row.toString());
}
}