[lake/paimon] Fix that tiering service for paimon missed timestamp type support

This closes  #1121.

Author: leonardBang

fluss-lake/fluss-lake-paimon/src/main/java/com/alibaba/fluss/lake/paimon/tiering/FlussRecordAsPaimonRow.java

@@ -18,41 +18,52 @@
 
 import com.alibaba.fluss.record.LogRecord;
 import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
 
 import org.apache.paimon.data.BinaryString;
 import org.apache.paimon.data.Decimal;
 import org.apache.paimon.data.InternalArray;
 import org.apache.paimon.data.InternalMap;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.types.DataType;
 import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
 
 import static com.alibaba.fluss.lake.paimon.utils.PaimonConversions.toRowKind;
+import static com.alibaba.fluss.utils.Preconditions.checkState;
 
 /** To wrap Fluss {@link LogRecord} as paimon {@link InternalRow}. */
 public class FlussRecordAsPaimonRow implements InternalRow {
 
+    // Lake table for paimon will append three system columns: __bucket, __offset,__timestamp
+    private static final int LAKE_PAIMON_SYSTEM_COLUMNS = 3;
+    private final RowType tableTowType;
     private final int bucket;
     private LogRecord logRecord;
     private int originRowFieldCount;
     private com.alibaba.fluss.row.InternalRow internalRow;
 
-    public FlussRecordAsPaimonRow(int bucket) {
+    public FlussRecordAsPaimonRow(int bucket, RowType tableTowType) {
         this.bucket = bucket;
+        this.tableTowType = tableTowType;
     }
 
     public void setFlussRecord(LogRecord logRecord) {
         this.logRecord = logRecord;
         this.internalRow = logRecord.getRow();
         this.originRowFieldCount = internalRow.getFieldCount();
+        checkState(
+                originRowFieldCount == tableTowType.getFieldCount() - LAKE_PAIMON_SYSTEM_COLUMNS,
+                "The paimon table fields count must equals to LogRecord's fields count.");
     }
 
     @Override
     public int getFieldCount() {
         return
         //  business (including partitions) + system (three system fields: bucket, offset,
         // timestamp)
-        originRowFieldCount + 3;
+        originRowFieldCount + LAKE_PAIMON_SYSTEM_COLUMNS;
     }
 
     @Override
@@ -142,13 +153,33 @@ public Timestamp getTimestamp(int pos, int precision) {
         if (pos == originRowFieldCount + 2) {
             return Timestamp.fromEpochMillis(logRecord.timestamp());
         }
-        if (TimestampLtz.isCompact(precision)) {
-            return Timestamp.fromEpochMillis(
-                    internalRow.getTimestampLtz(pos, precision).getEpochMillisecond());
-        } else {
-            TimestampLtz timestampLtz = internalRow.getTimestampLtz(pos, precision);
-            return Timestamp.fromEpochMillis(
-                    timestampLtz.getEpochMillisecond(), timestampLtz.getNanoOfMillisecond());
+
+        DataType paimonTimestampType = tableTowType.getTypeAt(pos);
+
+        switch (paimonTimestampType.getTypeRoot()) {
+            case TIMESTAMP_WITHOUT_TIME_ZONE:
+                if (TimestampNtz.isCompact(precision)) {
+                    return Timestamp.fromEpochMillis(
+                            internalRow.getTimestampNtz(pos, precision).getMillisecond());
+                } else {
+                    TimestampNtz timestampNtz = internalRow.getTimestampNtz(pos, precision);
+                    return Timestamp.fromEpochMillis(
+                            timestampNtz.getMillisecond(), timestampNtz.getNanoOfMillisecond());
+                }
+
+            case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+                if (TimestampLtz.isCompact(precision)) {
+                    return Timestamp.fromEpochMillis(
+                            internalRow.getTimestampLtz(pos, precision).getEpochMillisecond());
+                } else {
+                    TimestampLtz timestampLtz = internalRow.getTimestampLtz(pos, precision);
+                    return Timestamp.fromEpochMillis(
+                            timestampLtz.getEpochMillisecond(),
+                            timestampLtz.getNanoOfMillisecond());
+                }
+            default:
+                throw new UnsupportedOperationException(
+                        "Unsupported data type to get timestamp: " + paimonTimestampType);
         }
     }
 

fluss-lake/fluss-lake-paimon/src/main/java/com/alibaba/fluss/lake/paimon/tiering/RecordWriter.java

@@ -22,6 +22,7 @@
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.table.sink.CommitMessage;
 import org.apache.paimon.table.sink.TableWriteImpl;
+import org.apache.paimon.types.RowType;
 
 import javax.annotation.Nullable;
 
@@ -34,19 +35,23 @@
 public abstract class RecordWriter<T> implements AutoCloseable {
 
     protected final TableWriteImpl<T> tableWrite;
+    protected final RowType tableRowType;
     protected final int bucket;
     @Nullable protected final BinaryRow partition;
     protected final FlussRecordAsPaimonRow flussRecordAsPaimonRow;
 
     public RecordWriter(
             TableWriteImpl<T> tableWrite,
+            RowType tableRowType,
             TableBucket tableBucket,
             @Nullable String partition,
             List<String> partitionKeys) {
         this.tableWrite = tableWrite;
+        this.tableRowType = tableRowType;
         this.bucket = tableBucket.getBucket();
         this.partition = toPaimonPartitionBinaryRow(partitionKeys, partition);
-        this.flussRecordAsPaimonRow = new FlussRecordAsPaimonRow(tableBucket.getBucket());
+        this.flussRecordAsPaimonRow =
+                new FlussRecordAsPaimonRow(tableBucket.getBucket(), tableRowType);
     }
 
     public abstract void write(LogRecord record) throws Exception;

fluss-lake/fluss-lake-paimon/src/main/java/com/alibaba/fluss/lake/paimon/tiering/append/AppendOnlyWriter.java

@@ -43,6 +43,7 @@ public AppendOnlyWriter(
                 (TableWriteImpl<InternalRow>)
                         // todo: set ioManager to support write-buffer-spillable
                         fileStoreTable.newWrite(FLUSS_LAKE_TIERING_COMMIT_USER),
+                fileStoreTable.rowType(),
                 tableBucket,
                 partition,
                 partitionKeys); // Pass to parent

fluss-lake/fluss-lake-paimon/src/main/java/com/alibaba/fluss/lake/paimon/tiering/mergetree/MergeTreeWriter.java

@@ -44,7 +44,12 @@ public MergeTreeWriter(
             TableBucket tableBucket,
             @Nullable String partition,
             List<String> partitionKeys) {
-        super(createTableWrite(fileStoreTable), tableBucket, partition, partitionKeys);
+        super(
+                createTableWrite(fileStoreTable),
+                fileStoreTable.rowType(),
+                tableBucket,
+                partition,
+                partitionKeys);
         this.rowKeyExtractor = fileStoreTable.createRowKeyExtractor();
     }
 

fluss-lake/fluss-lake-paimon/src/test/java/com/alibaba/fluss/lake/paimon/tiering/FlussRecordAsPaimonRowTest.java

@@ -22,9 +22,11 @@
 import com.alibaba.fluss.row.Decimal;
 import com.alibaba.fluss.row.GenericRow;
 import com.alibaba.fluss.row.TimestampLtz;
+import com.alibaba.fluss.row.TimestampNtz;
 
 import org.apache.paimon.data.Timestamp;
 import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
 import org.junit.jupiter.api.Test;
 
 import java.math.BigDecimal;
@@ -42,8 +44,30 @@ class FlussRecordAsPaimonRowTest {
     @Test
     void testLogTableRecordAllTypes() {
         // Construct a FlussRecordAsPaimonRow instance
-        int bucket = 0;
-        FlussRecordAsPaimonRow flussRecordAsPaimonRow = new FlussRecordAsPaimonRow(bucket);
+        int tableBucket = 0;
+        RowType tableRowType =
+                RowType.of(
+                        new org.apache.paimon.types.BooleanType(),
+                        new org.apache.paimon.types.TinyIntType(),
+                        new org.apache.paimon.types.SmallIntType(),
+                        new org.apache.paimon.types.IntType(),
+                        new org.apache.paimon.types.BigIntType(),
+                        new org.apache.paimon.types.FloatType(),
+                        new org.apache.paimon.types.DoubleType(),
+                        new org.apache.paimon.types.VarCharType(),
+                        new org.apache.paimon.types.DecimalType(5, 2),
+                        new org.apache.paimon.types.DecimalType(20, 0),
+                        new org.apache.paimon.types.LocalZonedTimestampType(6),
+                        new org.apache.paimon.types.TimestampType(6),
+                        new org.apache.paimon.types.BinaryType(),
+                        new org.apache.paimon.types.VarCharType(),
+                        // append three system columns: __bucket, __offset,__timestamp
+                        new org.apache.paimon.types.IntType(),
+                        new org.apache.paimon.types.BigIntType(),
+                        new org.apache.paimon.types.LocalZonedTimestampType(3));
+
+        FlussRecordAsPaimonRow flussRecordAsPaimonRow =
+                new FlussRecordAsPaimonRow(tableBucket, tableRowType);
         long logOffset = 0;
         long timeStamp = System.currentTimeMillis();
         GenericRow genericRow = new GenericRow(14);
@@ -57,8 +81,8 @@ void testLogTableRecordAllTypes() {
         genericRow.setField(7, BinaryString.fromString("string"));
         genericRow.setField(8, Decimal.fromUnscaledLong(9, 5, 2));
         genericRow.setField(9, Decimal.fromBigDecimal(new BigDecimal(10), 20, 0));
-        genericRow.setField(10, TimestampLtz.fromEpochMillis(1698235273182L));
-        genericRow.setField(11, TimestampLtz.fromEpochMillis(1698235273182L, 45678));
+        genericRow.setField(10, TimestampLtz.fromEpochMillis(1698235273182L, 5678));
+        genericRow.setField(11, TimestampNtz.fromMillis(1698235273182L, 5678));
         genericRow.setField(12, new byte[] {1, 2, 3, 4});
         genericRow.setField(13, null);
         LogRecord logRecord = new GenericRecord(logOffset, timeStamp, APPEND_ONLY, genericRow);
@@ -77,15 +101,19 @@ void testLogTableRecordAllTypes() {
                 .isEqualTo(new BigDecimal("0.09"));
         assertThat(flussRecordAsPaimonRow.getDecimal(9, 20, 0).toBigDecimal())
                 .isEqualTo(new BigDecimal(10));
-        assertThat(flussRecordAsPaimonRow.getTimestamp(10, 3).getMillisecond())
+        assertThat(flussRecordAsPaimonRow.getTimestamp(10, 6).getMillisecond())
                 .isEqualTo(1698235273182L);
+        assertThat(flussRecordAsPaimonRow.getTimestamp(10, 6).getNanoOfMillisecond())
+                .isEqualTo(5678);
         assertThat(flussRecordAsPaimonRow.getTimestamp(11, 6).getMillisecond())
                 .isEqualTo(1698235273182L);
+        assertThat(flussRecordAsPaimonRow.getTimestamp(11, 6).getNanoOfMillisecond())
+                .isEqualTo(5678);
         assertThat(flussRecordAsPaimonRow.getBinary(12)).isEqualTo(new byte[] {1, 2, 3, 4});
         assertThat(flussRecordAsPaimonRow.isNullAt(13)).isTrue();
 
         // verify FlussRecordAsPaimonRow system columns (no partition fields, so indices stay same)
-        assertThat(flussRecordAsPaimonRow.getInt(14)).isEqualTo(bucket);
+        assertThat(flussRecordAsPaimonRow.getInt(14)).isEqualTo(tableBucket);
         assertThat(flussRecordAsPaimonRow.getLong(15)).isEqualTo(logOffset);
         assertThat(flussRecordAsPaimonRow.getLong(16)).isEqualTo(timeStamp);
         assertThat(flussRecordAsPaimonRow.getTimestamp(16, 4))
@@ -98,8 +126,16 @@ void testLogTableRecordAllTypes() {
 
     @Test
     void testPrimaryKeyTableRecord() {
-        int bucket = 0;
-        FlussRecordAsPaimonRow flussRecordAsPaimonRow = new FlussRecordAsPaimonRow(bucket);
+        int tableBucket = 0;
+        RowType tableRowType =
+                RowType.of(
+                        new org.apache.paimon.types.BooleanType(),
+                        // append three system columns: __bucket, __offset,__timestamp
+                        new org.apache.paimon.types.IntType(),
+                        new org.apache.paimon.types.BigIntType(),
+                        new org.apache.paimon.types.LocalZonedTimestampType(3));
+        FlussRecordAsPaimonRow flussRecordAsPaimonRow =
+                new FlussRecordAsPaimonRow(tableBucket, tableRowType);
         long logOffset = 0;
         long timeStamp = System.currentTimeMillis();
         GenericRow genericRow = new GenericRow(1);