[FLINK-38079] Add Pipeline support for DateType and TimeType

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/DateData.java

@@ -0,0 +1,78 @@
+/*
+ * 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.LocalDate;
+import java.util.Objects;
+
+/**
+ * An internal data structure representing data of {@link
+ * org.apache.flink.cdc.common.types.DateType}.
+ */
+public class DateData implements Comparable<DateData> {
+
+    private final int epochDay;
+
+    private DateData(int epochDay) {
+        this.epochDay = epochDay;
+    }
+
+    public static DateData fromEpochDay(int epochOfDay) {
+        return new DateData(epochOfDay);
+    }
+
+    public static DateData fromLocalDate(LocalDate date) {
+        return fromEpochDay((int) date.toEpochDay());
+    }
+
+    public static DateData fromIsoLocalDateString(String dateString) {
+        return fromLocalDate(LocalDate.parse(dateString));
+    }
+
+    public int toEpochDay() {
+        return epochDay;
+    }
+
+    public LocalDate toLocalDate() {
+        return LocalDate.ofEpochDay(epochDay);
+    }
+
+    public String toString() {
+        return toLocalDate().toString();
+    }
+
+    @Override
+    public final boolean equals(Object o) {
+        if (!(o instanceof DateData)) {
+            return false;
+        }
+
+        DateData dateData = (DateData) o;
+        return epochDay == dateData.epochDay;
+    }
+
+    @Override
+    public int compareTo(DateData other) {
+        return Long.compare(epochDay, other.epochDay);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(epochDay);
+    }
+}

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/RecordData.java

@@ -163,6 +163,12 @@ public interface RecordData {
      */
     RecordData getRow(int pos, int numFields);
 
+    /** Returns the Date data at the given position. */
+    DateData getDate(int pos);
+
+    /** Returns the Time data at the given position. */
+    TimeData getTime(int pos);
+
     /**
      * Creates an accessor for getting elements in an internal RecordData structure at the given
      * position.
@@ -197,9 +203,13 @@ static RecordData.FieldGetter createFieldGetter(DataType fieldType, int fieldPos
                 fieldGetter = record -> record.getShort(fieldPos);
                 break;
             case INTEGER:
+                fieldGetter = record -> record.getInt(fieldPos);
+                break;
             case DATE:
+                fieldGetter = record -> record.getDate(fieldPos);
+                break;
             case TIME_WITHOUT_TIME_ZONE:
-                fieldGetter = record -> record.getInt(fieldPos);
+                fieldGetter = record -> record.getTime(fieldPos);
                 break;
             case BIGINT:
                 fieldGetter = record -> record.getLong(fieldPos);

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/TimeData.java

@@ -0,0 +1,90 @@
+/*
+ * 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_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 fromNanoOfDay(long nanoOfDay) {
+        // millisOfDay should not exceed 86400000, which is safe to fit into INT.
+        return new TimeData((int) (nanoOfDay / MILLIS_TO_NANO));
+    }
+
+    public static TimeData fromLocalTime(LocalTime localTime) {
+        return fromNanoOfDay(localTime.toNanoOfDay());
+    }
+
+    public static TimeData fromIsoLocalTimeString(String timeString) {
+        return fromLocalTime(LocalTime.parse(timeString));
+    }
+
+    public int toMillisOfDay() {
+        return millisOfDay;
+    }
+
+    public LocalTime toLocalTime() {
+        return LocalTime.ofNanoOfDay((long) millisOfDay * MILLIS_TO_NANO);
+    }
+
+    public String toString() {
+        return toLocalTime().toString();
+    }
+
+    @Override
+    public final boolean equals(Object o) {
+        if (!(o instanceof TimeData)) {
+            return false;
+        }
+
+        TimeData timeData = (TimeData) o;
+        return millisOfDay == timeData.millisOfDay;
+    }
+
+    @Override
+    public int compareTo(TimeData other) {
+        return Long.compare(millisOfDay, other.millisOfDay);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(millisOfDay);
+    }
+}

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/binary/BinaryRecordData.java

@@ -19,11 +19,13 @@
 
 import org.apache.flink.cdc.common.annotation.Internal;
 import org.apache.flink.cdc.common.data.ArrayData;
+import org.apache.flink.cdc.common.data.DateData;
 import org.apache.flink.cdc.common.data.DecimalData;
 import org.apache.flink.cdc.common.data.LocalZonedTimestampData;
 import org.apache.flink.cdc.common.data.MapData;
 import org.apache.flink.cdc.common.data.RecordData;
 import org.apache.flink.cdc.common.data.StringData;
+import org.apache.flink.cdc.common.data.TimeData;
 import org.apache.flink.cdc.common.data.TimestampData;
 import org.apache.flink.cdc.common.data.ZonedTimestampData;
 import org.apache.flink.cdc.common.utils.Preconditions;
@@ -216,6 +218,18 @@ public RecordData getRow(int pos, int numFields) {
         return BinarySegmentUtils.readRecordData(segments, numFields, offset, getLong(pos));
     }
 
+    @Override
+    public DateData getDate(int pos) {
+        assertIndexIsValid(pos);
+        return DateData.fromEpochDay(getInt(pos));
+    }
+
+    @Override
+    public TimeData getTime(int pos) {
+        assertIndexIsValid(pos);
+        return TimeData.fromMillisOfDay(getInt(pos));
+    }
+
     /** The bit is 1 when the field is null. Default is 0. */
     @Override
     public boolean anyNull() {

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/utils/DataTypeUtils.java

@@ -18,10 +18,12 @@
 package org.apache.flink.cdc.common.types.utils;
 
 import org.apache.flink.cdc.common.data.ArrayData;
+import org.apache.flink.cdc.common.data.DateData;
 import org.apache.flink.cdc.common.data.DecimalData;
 import org.apache.flink.cdc.common.data.MapData;
 import org.apache.flink.cdc.common.data.RecordData;
 import org.apache.flink.cdc.common.data.StringData;
+import org.apache.flink.cdc.common.data.TimeData;
 import org.apache.flink.cdc.common.data.TimestampData;
 import org.apache.flink.cdc.common.data.ZonedTimestampData;
 import org.apache.flink.cdc.common.types.DataField;
@@ -58,9 +60,11 @@ public static Class<?> toInternalConversionClass(DataType type) {
             case SMALLINT:
                 return Short.class;
             case INTEGER:
+                return Integer.class;
             case DATE:
+                return DateData.class;
             case TIME_WITHOUT_TIME_ZONE:
-                return Integer.class;
+                return TimeData.class;
             case BIGINT:
                 return Long.class;
             case FLOAT:

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/DateTimeUtils.java

@@ -17,6 +17,9 @@
 
 package org.apache.flink.cdc.common.utils;
 
+import org.apache.flink.cdc.common.data.DateData;
+import org.apache.flink.cdc.common.data.TimeData;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -71,12 +74,12 @@ public class DateTimeUtils {
      * @param ts the timestamp in milliseconds.
      * @return the date in days.
      */
-    public static int timestampMillisToDate(long ts) {
-        int days = (int) (ts / MILLIS_PER_DAY);
+    public static DateData timestampMillisToDate(long ts) {
+        long days = ts / MILLIS_PER_DAY;
         if (days < 0) {
             days = days - 1;
         }
-        return days;
+        return DateData.fromEpochDay((int) days);
     }
 
     /**
@@ -85,8 +88,8 @@ public static int timestampMillisToDate(long ts) {
      * @param ts the timestamp in milliseconds.
      * @return the time in milliseconds.
      */
-    public static int timestampMillisToTime(long ts) {
-        return (int) (ts % MILLIS_PER_DAY);
+    public static TimeData timestampMillisToTime(long ts) {
+        return TimeData.fromMillisOfDay((int) (ts % MILLIS_PER_DAY));
     }
 
     // --------------------------------------------------------------------------------------------
@@ -103,12 +106,12 @@ public static int parseDate(String dateStr, String fromFormat) {
         return ymdToUnixDate(zdt.getYear(), zdt.getMonthValue(), zdt.getDayOfMonth());
     }
 
-    public static int parseDate(String dateStr, String fromFormat, String timezone) {
+    public static DateData parseDate(String dateStr, String fromFormat, String timezone) {
         long ts = internalParseTimestampMillis(dateStr, fromFormat, TimeZone.getTimeZone(timezone));
         ZoneId zoneId = ZoneId.of(timezone);
         Instant instant = Instant.ofEpochMilli(ts);
         ZonedDateTime zdt = ZonedDateTime.ofInstant(instant, zoneId);
-        return ymdToUnixDate(zdt.getYear(), zdt.getMonthValue(), zdt.getDayOfMonth());
+        return DateData.fromLocalDate(zdt.toLocalDate());
     }
 
     private static long internalParseTimestampMillis(String dateStr, String format, TimeZone tz) {