You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "proletarians (via GitHub)" <gi...@apache.org> on 2025/07/10 11:27:44 UTC
[PR] [FLINK-38079] Add Pipeline support for DateType and TimeType [flink-cdc]
proletarians opened a new pull request, #4060:
URL: https://github.com/apache/flink-cdc/pull/4060
Pipeline currently treats DateType and TimeType as plain INT, so date/time columns lose precision (TIME only to milliseconds, DATE range limited) and cannot be used with built-in temporal functions.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38079] Add Pipeline support for DateType and TimeType [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #4060:
URL: https://github.com/apache/flink-cdc/pull/4060#discussion_r2206771566
##########
flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java:
##########
@@ -135,11 +141,21 @@ void cleanup() {
System.setOut(standardOut);
}
+ private static Stream<Arguments> parameterProvider() {
+ return Stream.of(
+ Arguments.of(ValuesDataSink.SinkApi.SINK_FUNCTION, true),
+ Arguments.of(ValuesDataSink.SinkApi.SINK_FUNCTION, false),
+ Arguments.of(ValuesDataSink.SinkApi.SINK_V2, true),
+ Arguments.of(ValuesDataSink.SinkApi.SINK_V2, false));
+ }
+
/** This tests if we can append calculated columns based on existing columns. */
- @ParameterizedTest
- @EnumSource
- void testCalculatedColumns(ValuesDataSink.SinkApi sinkApi) throws Exception {
+ @ParameterizedTest(name = "API version: {0}, initializeMode: {1}")
+ @MethodSource(value = "parameterProvider")
+ void testCalculatedColumns(ValuesDataSink.SinkApi sinkApi, boolean initializeMode)
Review Comment:
What is `initializeMode`? Seems it doesn't affect test cases at all.
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaMergingUtils.java:
##########
@@ -85,6 +93,37 @@
*/
@PublicEvolving
public class SchemaMergingUtils {
+
Review Comment:
We may handle these changes in another PR and focus on supporting `DATE` and `TIME` formats here
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumSchemaDataTypeInference.java:
##########
@@ -48,6 +48,7 @@ public class DebeziumSchemaDataTypeInference implements SchemaDataTypeInference,
private static final long serialVersionUID = 1L;
+ public static final String DEBEZIUM_DATE_SCHEMA_NAME = "io.debezium.time.Date";
Review Comment:
Seems redundant as it's an alias of `io.debezium.time.Date.SCHEMA_NAME`?
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlFullTypesITCase.java:
##########
@@ -141,6 +143,7 @@ public void testMySql8JsonDataTypesWithUseLegacyJsonFormat() throws Throwable {
@Test
void testMysql57TimeDataTypes() throws Throwable {
+ UniqueDatabase usedDd = fullTypesMySql57Database;
Review Comment:
Please double check if these changes are necessary, and revert these to keep this PR minimum.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/writer/BinaryWriter.java:
##########
@@ -98,9 +105,21 @@ static void write(
writer.writeShort(pos, (short) o);
break;
case INTEGER:
+ writer.writeInt(pos, (int) o);
+ break;
case DATE:
+ if (o instanceof DateData) {
+ writer.writeDate(pos, (DateData) o);
+ } else {
+ writer.writeInt(pos, (int) o);
+ }
+ break;
Review Comment:
We may add some comments here, clarifying that the `writeInt` paths are kept for compatibility.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38079] Add Pipeline support for DateType and TimeType [flink-cdc]
Posted by "Mrart (via GitHub)" <gi...@apache.org>.
Mrart commented on PR #4060:
URL: https://github.com/apache/flink-cdc/pull/4060#issuecomment-3212789116
LGTM
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38079] Add Pipeline support for DateType and TimeType [flink-cdc]
Posted by "Mrart (via GitHub)" <gi...@apache.org>.
Mrart commented on code in PR #4060:
URL: https://github.com/apache/flink-cdc/pull/4060#discussion_r2292575242
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/TimeData.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.common.data;
+
+import java.time.LocalTime;
+import java.util.Objects;
+
+/**
+ * An internal data structure representing data of {@link
+ * org.apache.flink.cdc.common.types.TimeType}.
+ */
+public class TimeData implements Comparable<TimeData> {
+
+ private static final int SECONDS_TO_MILLIS = 1000;
+ private static final int MILLIS_TO_MICRO = 1000;
+ private static final int MILLIS_TO_NANO = 1_000_000;
+
+ private final int millisOfDay;
+
+ private TimeData(int millisOfDay) {
+ this.millisOfDay = millisOfDay;
+ }
+
+ public static TimeData fromSecondOfDay(int secondOfDay) {
+ return new TimeData(secondOfDay * SECONDS_TO_MILLIS);
+ }
+
+ public static TimeData fromMillisOfDay(int millisOfDay) {
+ return new TimeData(millisOfDay);
+ }
+
+ public static TimeData fromMicroOfDay(long microOfDay) {
+ return new TimeData((int) (microOfDay / MILLIS_TO_MICRO));
+ }
+
+ public static TimeData fromNanoOfDay(long nanoOfDay) {
+ // millisOfDay should not exceed 86400000, which is safe to fit into INT.
+ return new TimeData((int) (nanoOfDay / MILLIS_TO_NANO));
Review Comment:
Why only down to ms?not ns？
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38079] Add Pipeline support for DateType and TimeType [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #4060:
URL: https://github.com/apache/flink-cdc/pull/4060#discussion_r2292512293
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java:
##########
@@ -284,22 +286,22 @@ protected Object convertToFloat(Object dbzObj, Schema schema) {
}
protected Object convertToDate(Object dbzObj, Schema schema) {
- return (int) TemporalConversions.toLocalDate(dbzObj).toEpochDay();
+ return DateData.fromLocalDate(TemporalConversions.toLocalDate(dbzObj));
}
protected Object convertToTime(Object dbzObj, Schema schema) {
if (dbzObj instanceof Long) {
switch (schema.name()) {
case MicroTime.SCHEMA_NAME:
- return (int) ((long) dbzObj / 1000);
+ return TimeData.fromNanoOfDay((long) dbzObj * 1000);
Review Comment:
Makes sense, added `fromMicroOfDay` convenience init method.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38079] Add Pipeline support for DateType and TimeType [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #4060:
URL: https://github.com/apache/flink-cdc/pull/4060#discussion_r2290432100
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java:
##########
@@ -284,22 +286,22 @@ protected Object convertToFloat(Object dbzObj, Schema schema) {
}
protected Object convertToDate(Object dbzObj, Schema schema) {
- return (int) TemporalConversions.toLocalDate(dbzObj).toEpochDay();
+ return DateData.fromLocalDate(TemporalConversions.toLocalDate(dbzObj));
}
protected Object convertToTime(Object dbzObj, Schema schema) {
if (dbzObj instanceof Long) {
switch (schema.name()) {
case MicroTime.SCHEMA_NAME:
- return (int) ((long) dbzObj / 1000);
+ return TimeData.fromNanoOfDay((long) dbzObj * 1000);
Review Comment:
1 second = 1,000 milliSeconds = 1,000,000 microSeconds = 1,000,000,000 nanoSeconds.
So we have to use `fromMillisOfDay(x / 1000)`, which isn't better than `fromNanoOfDay(x * 1000)`
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38079] Add Pipeline support for DateType and TimeType [flink-cdc]
Posted by "Mrart (via GitHub)" <gi...@apache.org>.
Mrart commented on code in PR #4060:
URL: https://github.com/apache/flink-cdc/pull/4060#discussion_r2287234391
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java:
##########
@@ -284,22 +286,22 @@ protected Object convertToFloat(Object dbzObj, Schema schema) {
}
protected Object convertToDate(Object dbzObj, Schema schema) {
- return (int) TemporalConversions.toLocalDate(dbzObj).toEpochDay();
+ return DateData.fromLocalDate(TemporalConversions.toLocalDate(dbzObj));
}
protected Object convertToTime(Object dbzObj, Schema schema) {
if (dbzObj instanceof Long) {
switch (schema.name()) {
case MicroTime.SCHEMA_NAME:
- return (int) ((long) dbzObj / 1000);
+ return TimeData.fromNanoOfDay((long) dbzObj * 1000);
Review Comment:
TimeData implement From MillOfDay replace *1000 here?
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java:
##########
@@ -284,22 +286,22 @@ protected Object convertToFloat(Object dbzObj, Schema schema) {
}
protected Object convertToDate(Object dbzObj, Schema schema) {
- return (int) TemporalConversions.toLocalDate(dbzObj).toEpochDay();
+ return DateData.fromLocalDate(TemporalConversions.toLocalDate(dbzObj));
}
protected Object convertToTime(Object dbzObj, Schema schema) {
if (dbzObj instanceof Long) {
switch (schema.name()) {
case MicroTime.SCHEMA_NAME:
- return (int) ((long) dbzObj / 1000);
+ return TimeData.fromNanoOfDay((long) dbzObj * 1000);
case NanoTime.SCHEMA_NAME:
- return (int) ((long) dbzObj / 1000_000);
+ return TimeData.fromNanoOfDay((long) dbzObj);
}
} else if (dbzObj instanceof Integer) {
- return dbzObj;
+ return TimeData.fromNanoOfDay((int) dbzObj * 1_000_000L);
Review Comment:
TimeData implement fromSecondOfDay?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38079] Add Pipeline support for DateType and TimeType [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on PR #4060:
URL: https://github.com/apache/flink-cdc/pull/4060#issuecomment-3253293290
Just reverted an accidental change that breaks Postgres Pipeline CI, PTAL @lvyanquan
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38079] Add Pipeline support for DateType and TimeType [flink-cdc]
Posted by "Mrart (via GitHub)" <gi...@apache.org>.
Mrart commented on code in PR #4060:
URL: https://github.com/apache/flink-cdc/pull/4060#discussion_r2292502386
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java:
##########
@@ -284,22 +286,22 @@ protected Object convertToFloat(Object dbzObj, Schema schema) {
}
protected Object convertToDate(Object dbzObj, Schema schema) {
- return (int) TemporalConversions.toLocalDate(dbzObj).toEpochDay();
+ return DateData.fromLocalDate(TemporalConversions.toLocalDate(dbzObj));
}
protected Object convertToTime(Object dbzObj, Schema schema) {
if (dbzObj instanceof Long) {
switch (schema.name()) {
case MicroTime.SCHEMA_NAME:
- return (int) ((long) dbzObj / 1000);
+ return TimeData.fromNanoOfDay((long) dbzObj * 1000);
Review Comment:
It's possible to use these schemes in other places as well, so that you don't have to x*1000 everywhere, which will reduce the number of low-level errors when writing code
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38079] Add Pipeline support for DateType and TimeType [flink-cdc]
Posted by "Mrart (via GitHub)" <gi...@apache.org>.
Mrart commented on code in PR #4060:
URL: https://github.com/apache/flink-cdc/pull/4060#discussion_r2292796377
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/TimeData.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.common.data;
+
+import java.time.LocalTime;
+import java.util.Objects;
+
+/**
+ * An internal data structure representing data of {@link
+ * org.apache.flink.cdc.common.types.TimeType}.
+ */
+public class TimeData implements Comparable<TimeData> {
+
+ private static final int SECONDS_TO_MILLIS = 1000;
+ private static final int MILLIS_TO_MICRO = 1000;
+ private static final int MILLIS_TO_NANO = 1_000_000;
+
+ private final int millisOfDay;
+
+ private TimeData(int millisOfDay) {
+ this.millisOfDay = millisOfDay;
+ }
+
+ public static TimeData fromSecondOfDay(int secondOfDay) {
+ return new TimeData(secondOfDay * SECONDS_TO_MILLIS);
+ }
+
+ public static TimeData fromMillisOfDay(int millisOfDay) {
+ return new TimeData(millisOfDay);
+ }
+
+ public static TimeData fromMicroOfDay(long microOfDay) {
+ return new TimeData((int) (microOfDay / MILLIS_TO_MICRO));
+ }
+
+ public static TimeData fromNanoOfDay(long nanoOfDay) {
+ // millisOfDay should not exceed 86400000, which is safe to fit into INT.
+ return new TimeData((int) (nanoOfDay / MILLIS_TO_NANO));
Review Comment:
But I seem ok like this.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38079] Add Pipeline support for DateType and TimeType [flink-cdc]
Posted by "Mrart (via GitHub)" <gi...@apache.org>.
Mrart commented on PR #4060:
URL: https://github.com/apache/flink-cdc/pull/4060#issuecomment-3213165292
LGTM
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38079] Add Pipeline support for DateType and TimeType [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on PR #4060:
URL: https://github.com/apache/flink-cdc/pull/4060#issuecomment-3204610695
> DebeziumEventDeserializationSchema#convertToTime Should we also change the implementation of this class etc？
Updated.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38079] Add Pipeline support for DateType and TimeType [flink-cdc]
Posted by "Mrart (via GitHub)" <gi...@apache.org>.
Mrart commented on code in PR #4060:
URL: https://github.com/apache/flink-cdc/pull/4060#discussion_r2292779095
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/TimeData.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.common.data;
+
+import java.time.LocalTime;
+import java.util.Objects;
+
+/**
+ * An internal data structure representing data of {@link
+ * org.apache.flink.cdc.common.types.TimeType}.
+ */
+public class TimeData implements Comparable<TimeData> {
+
+ private static final int SECONDS_TO_MILLIS = 1000;
+ private static final int MILLIS_TO_MICRO = 1000;
+ private static final int MILLIS_TO_NANO = 1_000_000;
+
+ private final int millisOfDay;
+
+ private TimeData(int millisOfDay) {
+ this.millisOfDay = millisOfDay;
+ }
+
+ public static TimeData fromSecondOfDay(int secondOfDay) {
+ return new TimeData(secondOfDay * SECONDS_TO_MILLIS);
+ }
+
+ public static TimeData fromMillisOfDay(int millisOfDay) {
+ return new TimeData(millisOfDay);
+ }
+
+ public static TimeData fromMicroOfDay(long microOfDay) {
+ return new TimeData((int) (microOfDay / MILLIS_TO_MICRO));
+ }
+
+ public static TimeData fromNanoOfDay(long nanoOfDay) {
+ // millisOfDay should not exceed 86400000, which is safe to fit into INT.
+ return new TimeData((int) (nanoOfDay / MILLIS_TO_NANO));
Review Comment:
It's much better to have better timing accuracy for downstream processing where sorting is required
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38079] Add Pipeline support for DateType and TimeType [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan merged PR #4060:
URL: https://github.com/apache/flink-cdc/pull/4060
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38079] Add Pipeline support for DateType and TimeType [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #4060:
URL: https://github.com/apache/flink-cdc/pull/4060#discussion_r2290437408
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java:
##########
@@ -284,22 +286,22 @@ protected Object convertToFloat(Object dbzObj, Schema schema) {
}
protected Object convertToDate(Object dbzObj, Schema schema) {
- return (int) TemporalConversions.toLocalDate(dbzObj).toEpochDay();
+ return DateData.fromLocalDate(TemporalConversions.toLocalDate(dbzObj));
}
protected Object convertToTime(Object dbzObj, Schema schema) {
if (dbzObj instanceof Long) {
switch (schema.name()) {
case MicroTime.SCHEMA_NAME:
- return (int) ((long) dbzObj / 1000);
+ return TimeData.fromNanoOfDay((long) dbzObj * 1000);
case NanoTime.SCHEMA_NAME:
- return (int) ((long) dbzObj / 1000_000);
+ return TimeData.fromNanoOfDay((long) dbzObj);
}
} else if (dbzObj instanceof Integer) {
- return dbzObj;
+ return TimeData.fromNanoOfDay((int) dbzObj * 1_000_000L);
Review Comment:
Addressed, seems it should be `fromMillisOfDay`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38079] Add Pipeline support for DateType and TimeType [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #4060:
URL: https://github.com/apache/flink-cdc/pull/4060#discussion_r2292602545
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/TimeData.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.common.data;
+
+import java.time.LocalTime;
+import java.util.Objects;
+
+/**
+ * An internal data structure representing data of {@link
+ * org.apache.flink.cdc.common.types.TimeType}.
+ */
+public class TimeData implements Comparable<TimeData> {
+
+ private static final int SECONDS_TO_MILLIS = 1000;
+ private static final int MILLIS_TO_MICRO = 1000;
+ private static final int MILLIS_TO_NANO = 1_000_000;
+
+ private final int millisOfDay;
+
+ private TimeData(int millisOfDay) {
+ this.millisOfDay = millisOfDay;
+ }
+
+ public static TimeData fromSecondOfDay(int secondOfDay) {
+ return new TimeData(secondOfDay * SECONDS_TO_MILLIS);
+ }
+
+ public static TimeData fromMillisOfDay(int millisOfDay) {
+ return new TimeData(millisOfDay);
+ }
+
+ public static TimeData fromMicroOfDay(long microOfDay) {
+ return new TimeData((int) (microOfDay / MILLIS_TO_MICRO));
+ }
+
+ public static TimeData fromNanoOfDay(long nanoOfDay) {
+ // millisOfDay should not exceed 86400000, which is safe to fit into INT.
+ return new TimeData((int) (nanoOfDay / MILLIS_TO_NANO));
Review Comment:
We'd better keep the same memory layout as previous BinaryRecordData.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org