[Feature][Connector-V2] Suppor Time type in paimon connector (#8880)

Author: xiaochen-zhou

Diff for: docs/en/connector-v2/source/Paimon.md

@@ -73,6 +73,7 @@ The field data types currently supported by where conditions are as follows:
 * double
 * date
 * timestamp
+* time
 
 ### paimon.hadoop.conf [string]
 

Diff for: docs/zh/connector-v2/source/Paimon.md

@@ -74,7 +74,8 @@ Paimon 的 catalog uri，仅当 catalog_type 为 hive 时需要
 * float
 * double
 * date
-* timestamp
+* timestamp 
+* time
 
 ### paimon.hadoop.conf [string]
 

Diff for: seatunnel-common/src/main/java/org/apache/seatunnel/common/utils/TimeUtils.java

@@ -38,6 +38,10 @@ public static LocalTime parse(String time, Formatter formatter) {
         return LocalTime.parse(time, FORMATTER_MAP.get(formatter));
     }
 
+    public static LocalTime parse(String dateTime) {
+        return LocalTime.parse(dateTime, FORMATTER_MAP.get(matchTimeFormatter(dateTime)));
+    }
+
     public static final Pattern[] PATTERN_ARRAY =
             new Pattern[] {
                 Pattern.compile("\\d{2}:\\d{2}:\\d{2}"),

Diff for: seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/schema/UpdatedDataFields.java

@@ -43,6 +43,9 @@ public class UpdatedDataFields {
     private static final List<DataTypeRoot> TIMESTAMP_TYPES =
             Arrays.asList(DataTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE);
 
+    private static final List<DataTypeRoot> TIME_TYPES =
+            Arrays.asList(DataTypeRoot.TIME_WITHOUT_TIME_ZONE);
+
     public static ConvertAction canConvert(DataType oldType, DataType newType) {
         if (oldType.equalsIgnoreNullable(newType)) {
             return ConvertAction.CONVERT;
@@ -93,6 +96,14 @@ public static ConvertAction canConvert(DataType oldType, DataType newType) {
                     : ConvertAction.IGNORE;
         }
 
+        oldIdx = TIME_TYPES.indexOf(oldType.getTypeRoot());
+        newIdx = TIME_TYPES.indexOf(newType.getTypeRoot());
+        if (oldIdx >= 0 && newIdx >= 0) {
+            return DataTypeChecks.getPrecision(oldType) <= DataTypeChecks.getPrecision(newType)
+                    ? ConvertAction.CONVERT
+                    : ConvertAction.IGNORE;
+        }
+
         return ConvertAction.EXCEPTION;
     }
 

Diff for: seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/converter/SqlToPaimonPredicateConverter.java

@@ -17,6 +17,9 @@
 
 package org.apache.seatunnel.connectors.seatunnel.paimon.source.converter;
 
+import org.apache.seatunnel.common.utils.DateUtils;
+import org.apache.seatunnel.common.utils.TimeUtils;
+
 import org.apache.commons.lang3.StringUtils;
 import org.apache.paimon.data.BinaryString;
 import org.apache.paimon.data.Decimal;
@@ -266,8 +269,9 @@ private static Object convertValueByPaimonDataType(
                 case DOUBLE:
                     return Double.parseDouble(strValue);
                 case DATE:
-                    return DateTimeUtils.toInternal(
-                            org.apache.seatunnel.common.utils.DateUtils.parse(strValue));
+                    return DateTimeUtils.toInternal(DateUtils.parse(strValue));
+                case TIME_WITHOUT_TIME_ZONE:
+                    return DateTimeUtils.toInternal(TimeUtils.parse(strValue));
                 case TIMESTAMP_WITHOUT_TIME_ZONE:
                 case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
                     return Timestamp.fromLocalDateTime(

Diff for: seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/utils/RowConverter.java

@@ -53,6 +53,7 @@
 import java.math.BigDecimal;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
+import java.time.LocalTime;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -310,6 +311,10 @@ public static SeaTunnelRow convert(
                             convertArrayType(
                                     fieldName, paimonArray, seatunnelArray.getElementType());
                     break;
+                case TIME:
+                    int timeInt = rowData.getInt(i);
+                    objects[i] = DateTimeUtils.toLocalTime(timeInt);
+                    break;
                 case MAP:
                     MapType<?, ?> mapType = (MapType<?, ?>) fieldType;
                     InternalMap map = rowData.getMap(i);
@@ -434,6 +439,11 @@ public static InternalRow reconvert(
                     binaryWriter.writeTimestamp(
                             i, Timestamp.fromLocalDateTime(datetime), precision);
                     break;
+                case TIME:
+                    LocalTime time = (LocalTime) seaTunnelRow.getField(i);
+                    BinaryWriter.createValueSetter(DataTypes.TIME())
+                            .setValue(binaryWriter, i, DateTimeUtils.toInternal(time));
+                    break;
                 case MAP:
                     MapType<?, ?> mapType = (MapType<?, ?>) seaTunnelRowType.getFieldType(i);
                     SeaTunnelDataType<?> keyType = mapType.getKeyType();

Diff for: seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/utils/RowTypeConverter.java

@@ -178,8 +178,7 @@ public static Column convert(BasicTypeDefine<DataType> typeDefine) {
             case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
                 LocalZonedTimestampType localZonedTimestampType =
                         (LocalZonedTimestampType) dataType;
-                seaTunnelDataType =
-                        paimonToSeaTunnelTypeVisitor.visit((LocalZonedTimestampType) dataType);
+                seaTunnelDataType = paimonToSeaTunnelTypeVisitor.visit(localZonedTimestampType);
                 physicalColumnBuilder.scale(localZonedTimestampType.getPrecision());
                 break;
             case ARRAY:
@@ -228,6 +227,10 @@ public static RowType reconvert(SeaTunnelRowType seaTunnelRowType, TableSchema t
                 DataField dataField = SchemaUtil.getDataField(fields, fieldName);
                 dataType = new TimestampType(((TimestampType) dataField.type()).getPrecision());
             }
+            if (typeRoot.equals(DataTypeRoot.TIME_WITHOUT_TIME_ZONE)) {
+                DataField dataField = SchemaUtil.getDataField(fields, fieldName);
+                dataType = new TimeType(((TimeType) dataField.type()).getPrecision());
+            }
             DataField dataField = new DataField(i, fieldName, dataType);
             dataFields[i] = dataField;
         }
@@ -517,6 +520,11 @@ public SeaTunnelDataType<?> visit(TimestampType timestampType) {
             return LocalTimeType.LOCAL_DATE_TIME_TYPE;
         }
 
+        @Override
+        public SeaTunnelDataType<?> visit(TimeType timeType) {
+            return LocalTimeType.LOCAL_TIME_TYPE;
+        }
+
         @Override
         public SeaTunnelDataType<?> visit(LocalZonedTimestampType localZonedTimestampType) {
             return LocalTimeType.LOCAL_DATE_TIME_TYPE;

Diff for: seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/catalog/PaimonCatalogTest.java

@@ -181,7 +181,15 @@ public void before() {
                                         (Long) null,
                                         false,
                                         null,
-                                        "c_timestamp"));
+                                        "c_timestamp"))
+                        .column(
+                                PhysicalColumn.of(
+                                        "c_time",
+                                        LocalTimeType.LOCAL_TIME_TYPE,
+                                        (Long) null,
+                                        false,
+                                        null,
+                                        "c_time"));
     }
 
     @Test

Diff for: seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/schema/UpdatedDataFieldsTest.java

@@ -23,6 +23,7 @@
 import org.apache.paimon.types.FloatType;
 import org.apache.paimon.types.IntType;
 import org.apache.paimon.types.SmallIntType;
+import org.apache.paimon.types.TimeType;
 import org.apache.paimon.types.TimestampType;
 import org.apache.paimon.types.VarCharType;
 
@@ -98,4 +99,21 @@ public void testCanConvertTimestamp() {
 
         Assertions.assertEquals(UpdatedDataFields.ConvertAction.EXCEPTION, convertAction);
     }
+
+    @Test
+    public void testCanConvertTime() {
+        TimeType oldType = new TimeType(true, 3);
+        TimeType biggerLengthTimestamp = new TimeType(true, 5);
+        TimeType smallerLengthTimestamp = new TimeType(true, 2);
+        VarCharType varCharType = new VarCharType();
+
+        UpdatedDataFields.ConvertAction convertAction;
+        convertAction = UpdatedDataFields.canConvert(oldType, biggerLengthTimestamp);
+        Assertions.assertEquals(UpdatedDataFields.ConvertAction.CONVERT, convertAction);
+        convertAction = UpdatedDataFields.canConvert(oldType, smallerLengthTimestamp);
+        Assertions.assertEquals(UpdatedDataFields.ConvertAction.IGNORE, convertAction);
+        convertAction = UpdatedDataFields.canConvert(oldType, varCharType);
+
+        Assertions.assertEquals(UpdatedDataFields.ConvertAction.EXCEPTION, convertAction);
+    }
 }

Diff for: seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/converter/SqlToPaimonConverterTest.java

@@ -32,6 +32,7 @@
 import org.apache.paimon.types.IntType;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.types.SmallIntType;
+import org.apache.paimon.types.TimeType;
 import org.apache.paimon.types.TimestampType;
 import org.apache.paimon.types.TinyIntType;
 import org.apache.paimon.types.VarBinaryType;
@@ -46,6 +47,7 @@
 import java.math.BigDecimal;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
+import java.time.LocalTime;
 import java.util.Arrays;
 
 import static org.apache.seatunnel.connectors.seatunnel.paimon.source.converter.SqlToPaimonPredicateConverter.convertToPlainSelect;
@@ -77,7 +79,8 @@ public void setUp() {
                                 new DataField(9, "float_col", new FloatType()),
                                 new DataField(10, "double_col", new DoubleType()),
                                 new DataField(11, "date_col", new DateType()),
-                                new DataField(12, "timestamp_col", new TimestampType())));
+                                new DataField(12, "timestamp_col", new TimestampType()),
+                                new DataField(13, "time_col", new TimeType())));
 
         fieldNames = rowType.getFieldNames().toArray(new String[0]);
     }
@@ -97,7 +100,8 @@ public void testConvertSqlWhereToPaimonPredicate() {
                         + "float_col = 5.5 AND "
                         + "double_col = 6.6 AND "
                         + "date_col = '2022-01-01' AND "
-                        + "timestamp_col = '2022-01-01T12:00:00.123'";
+                        + "timestamp_col = '2022-01-01T12:00:00.123' AND "
+                        + "time_col = '12:00:00.123'";
 
         PlainSelect plainSelect = convertToPlainSelect(query);
         Predicate predicate =
@@ -125,7 +129,9 @@ public void testConvertSqlWhereToPaimonPredicate() {
                         builder.equal(
                                 12,
                                 Timestamp.fromLocalDateTime(
-                                        LocalDateTime.parse("2022-01-01T12:00:00.123"))));
+                                        LocalDateTime.parse("2022-01-01T12:00:00.123"))),
+                        builder.equal(
+                                13, DateTimeUtils.toInternal(LocalTime.parse("12:00:00.123"))));
 
         assertEquals(expectedPredicate.toString(), predicate.toString());
     }

Diff for: seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/utils/RowConverterTest.java

@@ -57,6 +57,7 @@
 import java.math.BigDecimal;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
+import java.time.LocalTime;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
@@ -90,7 +91,8 @@ public class RowConverterTest {
         "c_date",
         "c_timestamp",
         "c_map",
-        "c_array"
+        "c_array",
+        "c_time"
     };
 
     public static final SeaTunnelDataType<?>[] seaTunnelDataTypes = {
@@ -107,7 +109,8 @@ public class RowConverterTest {
         LocalTimeType.LOCAL_DATE_TYPE,
         LocalTimeType.LOCAL_DATE_TIME_TYPE,
         new MapType<>(BasicType.STRING_TYPE, BasicType.STRING_TYPE),
-        ArrayType.STRING_ARRAY_TYPE
+        ArrayType.STRING_ARRAY_TYPE,
+        LocalTimeType.LOCAL_TIME_TYPE
     };
 
     public static final List<String> KEY_NAME_LIST = Arrays.asList("c_tinyint");
@@ -129,7 +132,8 @@ public TableSchema getTableSchema(int decimalPrecision, int decimalScale) {
                             DataTypes.DATE(),
                             DataTypes.TIMESTAMP(),
                             DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()),
-                            DataTypes.ARRAY(DataTypes.STRING())
+                            DataTypes.ARRAY(DataTypes.STRING()),
+                            DataTypes.TIME()
                         },
                         new String[] {
                             "c_tinyint",
@@ -145,7 +149,8 @@ public TableSchema getTableSchema(int decimalPrecision, int decimalScale) {
                             "c_date",
                             "c_timestamp",
                             "c_map",
-                            "c_array"
+                            "c_array",
+                            "c_time",
                         });
 
         return new TableSchema(
@@ -172,11 +177,12 @@ public void generateTestData() {
         byte[] bytes = new byte[] {1, 2, 3, 4};
         boolean booleanValue = false;
         LocalDate date = LocalDate.of(1996, 3, 16);
+        LocalTime time = LocalTime.of(12, 0, 0);
         LocalDateTime timestamp = LocalDateTime.of(1996, 3, 16, 4, 16, 20);
         Map<String, String> map = new HashMap<>();
         map.put("name", "paimon");
         String[] strings = new String[] {"paimon", "seatunnel"};
-        Object[] objects = new Object[14];
+        Object[] objects = new Object[15];
         objects[0] = tinyint;
         objects[1] = smallint;
         objects[2] = intNum;
@@ -191,8 +197,9 @@ public void generateTestData() {
         objects[11] = timestamp;
         objects[12] = map;
         objects[13] = strings;
+        objects[14] = time;
         seaTunnelRow = new SeaTunnelRow(objects);
-        BinaryRow binaryRow = new BinaryRow(14);
+        BinaryRow binaryRow = new BinaryRow(15);
         BinaryRowWriter binaryRowWriter = new BinaryRowWriter(binaryRow);
         binaryRowWriter.writeByte(0, tinyint);
         binaryRowWriter.writeShort(1, smallint);
@@ -234,6 +241,7 @@ public void generateTestData() {
         binaryArrayWriter2.complete();
         binaryRowWriter.writeArray(
                 13, binaryArray2, new InternalArraySerializer(DataTypes.STRING()));
+        binaryRowWriter.writeInt(14, DateTimeUtils.toInternal(time));
         internalRow = binaryRow;
     }
 

Diff for: seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/utils/RowTypeConverterTest.java

@@ -71,6 +71,7 @@ public class RowTypeConverterTest {
                         DataTypes.BOOLEAN(),
                         DataTypes.DATE(),
                         DataTypes.TIMESTAMP(),
+                        DataTypes.TIME(),
                         DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()),
                         DataTypes.ARRAY(DataTypes.STRING())
                     },
@@ -87,6 +88,7 @@ public class RowTypeConverterTest {
                         "c_boolean",
                         "c_date",
                         "c_timestamp",
+                        "c_time",
                         "c_map",
                         "c_array"
                     });
@@ -110,6 +112,7 @@ public void before() {
                             "c_boolean",
                             "c_date",
                             "c_timestamp",
+                            "c_time",
                             "c_map",
                             "c_array"
                         },
@@ -126,6 +129,7 @@ public void before() {
                             BasicType.BOOLEAN_TYPE,
                             LocalTimeType.LOCAL_DATE_TYPE,
                             LocalTimeType.LOCAL_DATE_TIME_TYPE,
+                            LocalTimeType.LOCAL_TIME_TYPE,
                             new MapType<>(BasicType.STRING_TYPE, BasicType.STRING_TYPE),
                             ArrayType.STRING_ARRAY_TYPE
                         });
@@ -149,9 +153,10 @@ public void before() {
                         new DataField(9, "c_boolean", DataTypes.BOOLEAN()),
                         new DataField(10, "c_date", DataTypes.DATE()),
                         new DataField(11, "c_timestamp", DataTypes.TIMESTAMP(6)),
+                        new DataField(12, "c_time", DataTypes.TIME()),
                         new DataField(
-                                12, "c_map", DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING())),
-                        new DataField(13, "c_array", DataTypes.ARRAY(DataTypes.STRING())));
+                                13, "c_map", DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING())),
+                        new DataField(14, "c_array", DataTypes.ARRAY(DataTypes.STRING())));
 
         tableSchema =
                 new TableSchema(

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paimon-e2e/src/test/java/org/apache/seatunnel/e2e/connector/paimon/PaimonRecordWithFullType.java

@@ -46,4 +46,5 @@ public class PaimonRecordWithFullType {
     public BinaryString c_bytes;
     public int c_date;
     public Timestamp c_timestamp;
+    public int c_time;
 }

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paimon-e2e/src/test/java/org/apache/seatunnel/e2e/connector/paimon/PaimonSinkDynamicBucketIT.java

@@ -481,7 +481,8 @@ private List<PaimonRecordWithFullType> loadPaimonDataWithFullType(FileStoreTable
                                         row.getDecimal(10, 30, 8),
                                         row.getString(11),
                                         row.getInt(12),
-                                        row.getTimestamp(13, 6));
+                                        row.getTimestamp(13, 6),
+                                        row.getInt(14));
                         result.add(paimonRecordWithFullType);
                     });
         } catch (IOException e) {

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paimon-e2e/src/test/java/org/apache/seatunnel/e2e/connector/paimon/PaimonStreamReadIT.java

@@ -141,7 +141,8 @@ protected List<PaimonRecordWithFullType> loadPaimonDataWithFullType(
                                         row.getDecimal(10, 30, 8),
                                         row.getString(11),
                                         row.getInt(12),
-                                        row.getTimestamp(13, 6));
+                                        row.getTimestamp(13, 6),
+                                        row.getInt(14));
                         result.add(paimonRecordWithFullType);
                     });
         } catch (IOException e) {

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paimon-e2e/src/test/resources/fake_to_dynamic_bucket_paimon_case1.conf

@@ -40,6 +40,7 @@ source {
         c_bytes = bytes
         c_date = date
         c_timestamp = timestamp
+        c_time = time
       }
       primaryKey {
           name = "pk_id"