yuxiqian commented on code in PR #4530:
URL: https://github.com/apache/flink-cdc/pull/4530#discussion_r4068853723
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/AbstractBinaryWriter.java:
##########
@@ -236,7 +236,11 @@ public void writeDate(int pos, DateData value) {
@Override
public void writeTime(int pos, TimeData value, int precision) {
- writeInt(pos, value.toMillisOfDay());
+ if (precision <= 3) {
+ writeInt(pos, value.toMillisOfDay());
+ } else {
+ writeLong(pos, Long.MIN_VALUE | value.toNanoOfDay());
+ }
Review Comment:
If you refer to the writeTimestamp method, you may see that the precision
for temporal types are not enforced, and we won't truncate over-precision
digits or use different memory layouts depending on precisions. I would prefer
follow the same way and avoid modifying too much public API.
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/binary/BinaryArrayData.java:
##########
@@ -226,6 +229,21 @@ public int getInt(int pos) {
return BinarySegmentUtils.getInt(segments, getElementOffset(pos, 4));
}
+ @Override
+ public TimeData getTime(int pos, int precision) {
+ assertIndexIsValid(pos);
+ if (precision <= 3) {
Review Comment:
It is possible not to use different memory layout? Seems fragile if
precision were lost in the pipeline.
##########
flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java:
##########
@@ -3072,7 +3072,7 @@ void testDateAndTimeCastingFunctions() throws Exception {
assertThat(outputEvents)
.containsExactlyInAnyOrder(
"CreateTableEvent{tableId=default_namespace.default_schema.my_table,
schema=columns={`id` INT NOT NULL,`date_0` DATE,`time_0` TIME(0),`time_3`
TIME(3),`time_6` TIME(6),`time_9` TIME(9),`date_0_str` STRING,`time_0_str`
STRING,`time_3_str` STRING,`time_6_str` STRING,`time_9_str` STRING},
primaryKeys=id, options=()}",
-
"DataChangeEvent{tableId=default_namespace.default_schema.my_table, before=[],
after=[1, 1999-12-31, 21:48:25, 21:48:25.123, 21:48:25.123, 21:48:25.123,
1999-12-31, 21:48:25, 21:48:25.123, 21:48:25.123, 21:48:25.123], op=INSERT,
meta=()}",
+
"DataChangeEvent{tableId=default_namespace.default_schema.my_table, before=[],
after=[1, 1999-12-31, 21:48:25, 21:48:25.123, 21:48:25.123456,
21:48:25.123456789, 1999-12-31, 21:48:25, 21:48:25.123, 21:48:25.123456,
21:48:25.123456789], op=INSERT, meta=()}",
Review Comment:
Please also add tests to verify how transform / routing operators handle
TIME with different precisions.
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/RecordData.java:
##########
@@ -170,6 +170,16 @@ public interface RecordData {
/** Returns the Time data at the given position. */
TimeData getTime(int pos);
+ /**
+ * Returns the Time data at the given position using its declared
precision.
+ *
+ * <p>The default implementation preserves compatibility with record
implementations whose
+ * representation is independent of precision.
+ */
+ default TimeData getTime(int pos, int precision) {
+ return getTime(pos);
+ }
Review Comment:
It's concerning to modify public API. Shall we use the same memory layout
for all precisions, so this would be unnecessary?
--
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]
